appview/repo: add new spindles to consumer as they are ingested

Signed-off-by: oppiliappan <[e-mail]>

appview/ingester.go

···
 	return nil
 }
 
-func ingestSpindleMember(d *db.DbWrapper, e *models.Event, enforcer *rbac.Enforcer) error {
+func ingestSpindleMember(_ *db.DbWrapper, e *models.Event, enforcer *rbac.Enforcer) error {
 	did := e.Did
 	var err error
 
···
 		}
 
 		// this is a special record whose rkey is the instance of the spindle itself
-		domain := e.Commit.RKey
+		instance := e.Commit.RKey
 
-		owner, err := fetchOwner(context.TODO(), domain, true)
+		owner, err := fetchOwner(context.TODO(), instance, dev)
 		if err != nil {
-			log.Printf("failed to verify owner of %s: %w", domain, err)
+			log.Printf("failed to verify owner of %s: %s", instance, err)
 			return err
 		}
 
 		// verify that the spindle owner points back to this did
 		if owner != did {
-			log.Printf("incorrect owner for domain: %s, %s != %s", domain, owner, did)
+			log.Printf("incorrect owner for domain: %s, %s != %s", instance, owner, did)
 			return err
 		}
 
 		// mark this spindle as registered
+		ddb, ok := d.Execer.(*db.DB)
+		if !ok {
+			return fmt.Errorf("failed to index profile record, invalid db cast")
+		}
+
+		_, err = db.VerifySpindle(
+			ddb,
+			db.FilterEq("owner", did),
+			db.FilterEq("instance", instance),
+		)
+
+		return err
 	}
 
 	return nil

appview/repo/repo.go

···
 package repo
 
 import (
+	"context"
 	"database/sql"
 	"encoding/json"
 	"errors"
···
 	"tangled.sh/tangled.sh/core/appview/pages/markup"
 	"tangled.sh/tangled.sh/core/appview/pages/repoinfo"
 	"tangled.sh/tangled.sh/core/appview/reporesolver"
+	"tangled.sh/tangled.sh/core/eventconsumer"
 	"tangled.sh/tangled.sh/core/knotclient"
 	"tangled.sh/tangled.sh/core/patchutil"
 	"tangled.sh/tangled.sh/core/rbac"
···
 )
 
 type Repo struct {
-	repoResolver *reporesolver.RepoResolver
-	idResolver   *idresolver.Resolver
-	config       *config.Config
-	oauth        *oauth.OAuth
-	pages        *pages.Pages
-	db           *db.DB
-	enforcer     *rbac.Enforcer
-	posthog      posthog.Client
+	repoResolver  *reporesolver.RepoResolver
+	idResolver    *idresolver.Resolver
+	config        *config.Config
+	oauth         *oauth.OAuth
+	pages         *pages.Pages
+	spindlestream *eventconsumer.Consumer
+	db            *db.DB
+	enforcer      *rbac.Enforcer
+	posthog       posthog.Client
 }
 
 func New(
 	oauth *oauth.OAuth,
 	repoResolver *reporesolver.RepoResolver,
 	pages *pages.Pages,
+	spindlestream *eventconsumer.Consumer,
 	idResolver *idresolver.Resolver,
 	db *db.DB,
 	config *config.Config,
···
 	enforcer *rbac.Enforcer,
 ) *Repo {
 	return &Repo{oauth: oauth,
-		repoResolver: repoResolver,
-		pages:        pages,
-		idResolver:   idResolver,
-		config:       config,
-		db:           db,
-		posthog:      posthog,
-		enforcer:     enforcer,
+		repoResolver:  repoResolver,
+		pages:         pages,
+		idResolver:    idResolver,
+		config:        config,
+		spindlestream: spindlestream,
+		db:            db,
+		posthog:       posthog,
+		enforcer:      enforcer,
 	}
 }
 
···
 		rp.pages.Notice(w, "repo-notice", "Failed to configure spindle, unable to save to PDS.")
 		return
 	}
+
+	// add this spindle to spindle stream
+	rp.spindlestream.AddSource(
+		context.Background(),
+		eventconsumer.NewSpindleSource(newSpindle),
+	)
 
 	w.Write(fmt.Append(nil, "spindle set to: ", newSpindle))
 }

appview/state/router.go

···
 }
 
 func (s *State) RepoRouter(mw *middleware.Middleware) http.Handler {
-	repo := repo.New(s.oauth, s.repoResolver, s.pages, s.idResolver, s.db, s.config, s.posthog, s.enforcer)
+	repo := repo.New(s.oauth, s.repoResolver, s.pages, s.spindlestream, s.idResolver, s.db, s.config, s.posthog, s.enforcer)
 	return repo.Router(mw)
 }