func (w *Watcher) SyncWatcher(ctx context.Context, in *v1.SyncWatcherRequest) (*v1.SyncWatcherResponse, error) {
	id := in.Id
	fmt.Printf("syncing watcher: %v", id)
	// fetch watcher index
	watcher, err := w.store.Get(id)
	if err != nil {
		return nil, fmt.Errorf("error synchronising watcher %v, error: %v", id, err)
	}
	source, err := sources.GetSource(watcher.Spec.Source.Parser.String())
	if err != nil {
		return nil, fmt.Errorf("failed to get source %v, error: %v", watcher.Spec.Source.Parser, err)
	}
	handler, err := source.NewBuilder().Build()
	if err != nil {
		return nil, fmt.Errorf("failed to build handler, error: %v", err)
	}
	manager, err := handler.Parse(ctx, watcher.Spec.Source)

	if err != nil {
		return nil, fmt.Errorf("failed to parse source: %s", watcher.Spec.Source.Url)
	}

	difference := manager.Difference(watcher.Spec.Index)

	// 1. start collecting new documents - async go routine?
	// 2. once completed we update index with successfully collected documents - needs to do ith a mutex


	log.Infof(ctx, "[Source:%v]: found %v new documents", watcher.Spec.Source.Url, difference.GetIndex().Count)

	difference.Collect(watcher)

	return &v1.SyncWatcherResponse{Sync: &v1.SyncDetails{
		Id:     id,
		Name:   watcher.Metadata.Name,
		Status: v1.Status_SUCCESS,
	}}, nil
}
// SyncWatcher is responsible for discovering outstanding documents available for collection and managing the
// collection process. General workflow is as follows:
// 1. fetch current index
// 2. discover latest content from given source
// 3. compare difference between current and latest documents
// 4. synchronise/collect new documents via the collector service
// 5. update index with new state based on results of synchronisation status
SyncWatcher is responsible for discovering outstanding documents available for collection and managing the
collection process. General workflow is as follows:
1. fetch current index
2. discover latest content from given source
3. compare difference between current and latest documents
4. synchronise/collect new documents via the collector service
5. update index with new state based on results of synchronisation status
// SyncWatcher is responsible for discovering outstanding documents available for collection and managing the // collection process . General workflow is as follows : // 1 . fetch current index // 2 . discover latest content from given source // 3 . compare difference between current and latest documents // 4 . synchronise / collect new documents via the collector service // 5 . update index with new state based on results of synchronisation status
func (w *Watcher) SyncWatcher(ctx context.Context, in *v1.SyncWatcherRequest) (*v1.SyncWatcherResponse, error) {
	id := in.Id
	fmt.Printf("syncing watcher: %v", id)
	watcher, err := w.store.Get(id)
	if err != nil {
		return nil, fmt.Errorf("error synchronising watcher %v, error: %v", id, err)
	}
	source, err := sources.GetSource(watcher.Spec.Source.Parser.String())
	if err != nil {
		return nil, fmt.Errorf("failed to get source %v, error: %v", watcher.Spec.Source.Parser, err)
	}
	handler, err := source.NewBuilder().Build()
	if err != nil {
		return nil, fmt.Errorf("failed to build handler, error: %v", err)
	}
	manager, err := handler.Parse(ctx, watcher.Spec.Source)
	if err != nil {
		return nil, fmt.Errorf("failed to parse source: %s", watcher.Spec.Source.Url)
	}
	difference := manager.Difference(watcher.Spec.Index)
	log.Infof(ctx, "[Source:%v]: found %v new documents", watcher.Spec.Source.Url, difference.GetIndex().Count)
	difference.Collect(watcher)
	return &v1.SyncWatcherResponse{Sync: &v1.SyncDetails{
		Id:     id,
		Name:   watcher.Metadata.Name,
		Status: v1.Status_SUCCESS,
	}}, nil
}
func ( w * Watcher ) SyncWatcher ( ctx context . Context , in * v1 . SyncWatcherRequest ) ( * v1 . SyncWatcherResponse , error ) { id := in . Id fmt . Printf ( " " , id ) watcher , err := w . store . Get ( id ) if err != nil { return nil , fmt . Errorf ( " " , id , err ) } source , err := sources . GetSource ( watcher . Spec . Source . Parser . String ( ) ) if err != nil { return nil , fmt . Errorf ( " " , watcher . Spec . Source . Parser , err ) } handler , err := source . NewBuilder ( ) . Build ( ) if err != nil { return nil , fmt . Errorf ( " " , err ) } manager , err := handler . Parse ( ctx , watcher . Spec . Source ) if err != nil { return nil , fmt . Errorf ( " " , watcher . Spec . Source . Url ) } difference := manager . Difference ( watcher . Spec . Index ) log . Infof ( ctx , " " , watcher . Spec . Source . Url , difference . GetIndex ( ) . Count ) difference . Collect ( watcher ) return & v1 . SyncWatcherResponse { Sync : & v1 . SyncDetails { Id : id , Name : watcher . Metadata . Name , Status : v1 . Status_SUCCESS , } } , nil }
SyncWatcher is responsible for discovering outstanding documents available for collection and managing the
collection process.
SyncWatcher is responsible for discovering outstanding documents available for collection and managing the collection process .
// fetch watcher index // 1. start collecting new documents - async go routine? // 2. once completed we update index with successfully collected documents - needs to do ith a mutex