finish backend refactor

backend/backend.go

···
 	"fmt"
 	"strings"
 	"sync"
+	"time"
 
+	"github.com/bluesky-social/indigo/api/atproto"
+	"github.com/bluesky-social/indigo/api/bsky"
 	"github.com/bluesky-social/indigo/atproto/syntax"
 	"github.com/bluesky-social/indigo/util"
+	"github.com/bluesky-social/indigo/xrpc"
 	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
···
 	. "github.com/whyrusleeping/konbini/models"
 	"github.com/whyrusleeping/market/models"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 	"gorm.io/gorm/logger"
 )
 
···
 	pgx     *pgxpool.Pool
 	tracker RecordTracker
 
+	client *xrpc.Client
+
+	mydid  string
+	myrepo *models.Repo
+
 	relevantDids map[string]bool
 	rdLk         sync.Mutex
 
···
 }
 
 // NewPostgresBackend creates a new PostgresBackend
-func NewPostgresBackend(db *gorm.DB, pgx *pgxpool.Pool, tracker RecordTracker) (*PostgresBackend, error) {
-	rc, err := lru.New2Q[string, *Repo](1_000_000)
-	if err != nil {
-		return nil, err
-	}
-	pc, err := lru.New2Q[string, cachedPostInfo](1_000_000)
-	if err != nil {
-		return nil, err
-	}
-	revc, err := lru.New2Q[uint, string](1_000_000)
-	if err != nil {
-		return nil, err
-	}
+func NewPostgresBackend(mydid string, db *gorm.DB, pgx *pgxpool.Pool, client *xrpc.Client, tracker RecordTracker) (*PostgresBackend, error) {
+	rc, _ := lru.New2Q[string, *Repo](1_000_000)
+	pc, _ := lru.New2Q[string, cachedPostInfo](1_000_000)
+	revc, _ := lru.New2Q[uint, string](1_000_000)
 
-	return &PostgresBackend{
+	b := &PostgresBackend{
+		client:        client,
+		mydid:         mydid,
 		db:            db,
 		pgx:           pgx,
 		tracker:       tracker,
···
 		repoCache:     rc,
 		postInfoCache: pc,
 		revCache:      revc,
-	}, nil
+	}
+
+	r, err := b.GetOrCreateRepo(context.TODO(), mydid)
+	if err != nil {
+		return nil, err
+	}
+
+	b.myrepo = r
+	return b, nil
 }
 
 // TrackMissingRecord implements the RecordTracker interface
···
 
 // DidToID converts a DID to a database ID
 func (b *PostgresBackend) DidToID(ctx context.Context, did string) (uint, error) {
-	r, err := b.getOrCreateRepo(ctx, did)
+	r, err := b.GetOrCreateRepo(ctx, did)
 	if err != nil {
 		return 0, err
 	}
 	return r.ID, nil
 }
 
-func (b *PostgresBackend) getOrCreateRepo(ctx context.Context, did string) (*Repo, error) {
+func (b *PostgresBackend) GetOrCreateRepo(ctx context.Context, did string) (*Repo, error) {
 	r, ok := b.repoCache.Get(did)
 	if !ok {
 		b.reposLk.Lock()
···
 	return r, nil
 }
 
-func (b *PostgresBackend) getOrCreateList(ctx context.Context, uri string) (*List, error) {
+func (b *PostgresBackend) GetOrCreateList(ctx context.Context, uri string) (*List, error) {
 	puri, err := util.ParseAtUri(uri)
 	if err != nil {
 		return nil, err
 	}
 
-	r, err := b.getOrCreateRepo(ctx, puri.Did)
+	r, err := b.GetOrCreateRepo(ctx, puri.Did)
 	if err != nil {
 		return nil, err
 	}
···
 		return nil, err
 	}
 
-	r, err := b.getOrCreateRepo(ctx, puri.Did)
+	r, err := b.GetOrCreateRepo(ctx, puri.Did)
 	if err != nil {
 		return nil, err
 	}
···
 	return post, nil
 }
 
-func (b *PostgresBackend) getPostByUri(ctx context.Context, uri string, fields string) (*Post, error) {
+func (b *PostgresBackend) GetPostByUri(ctx context.Context, uri string, fields string) (*Post, error) {
 	puri, err := util.ParseAtUri(uri)
 	if err != nil {
 		return nil, err
 	}
 
-	r, err := b.getOrCreateRepo(ctx, puri.Did)
+	r, err := b.GetOrCreateRepo(ctx, puri.Did)
 	if err != nil {
 		return nil, err
 	}
···
 	return rev, nil
 }
 
-func (b *PostgresBackend) addRelevantDid(did string) {
+func (b *PostgresBackend) AddRelevantDid(did string) {
 	b.rdLk.Lock()
 	defer b.rdLk.Unlock()
 	b.relevantDids[did] = true
 }
 
-func (b *PostgresBackend) didIsRelevant(did string) bool {
+func (b *PostgresBackend) DidIsRelevant(did string) bool {
 	b.rdLk.Lock()
 	defer b.rdLk.Unlock()
 	return b.relevantDids[did]
···
 func (b *PostgresBackend) anyRelevantIdents(idents ...string) bool {
 	for _, id := range idents {
 		if strings.HasPrefix(id, "did:") {
-			if b.didIsRelevant(id) {
+			if b.DidIsRelevant(id) {
 				return true
 			}
 		} else if strings.HasPrefix(id, "at://") {
···
 				continue
 			}
 
-			if b.didIsRelevant(puri.Authority().String()) {
+			if b.DidIsRelevant(puri.Authority().String()) {
 				return true
 			}
 		}
···
 	return false
 }
 
-func (b *PostgresBackend) getRepoByID(ctx context.Context, id uint) (*models.Repo, error) {
+func (b *PostgresBackend) GetRelevantDids() []string {
+	b.rdLk.Lock()
+	var out []string
+	for k := range b.relevantDids {
+		out = append(out, k)
+	}
+	b.rdLk.Unlock()
+	return out
+}
+
+func (b *PostgresBackend) GetRepoByID(ctx context.Context, id uint) (*models.Repo, error) {
 	var r models.Repo
 	if err := b.db.Find(&r, "id = ?", id).Error; err != nil {
 		return nil, err
···
 
 	return false, nil
 }
+
+func (b *PostgresBackend) LoadRelevantDids() error {
+	ctx := context.TODO()
+
+	if err := b.ensureFollowsScraped(ctx, b.mydid); err != nil {
+		return fmt.Errorf("failed to scrape follows: %w", err)
+	}
+
+	r, err := b.GetOrCreateRepo(ctx, b.mydid)
+	if err != nil {
+		return err
+	}
+
+	var dids []string
+	if err := b.db.Raw("select did from follows left join repos on follows.subject = repos.id where follows.author = ?", r.ID).Scan(&dids).Error; err != nil {
+		return err
+	}
+
+	b.relevantDids[b.mydid] = true
+	for _, d := range dids {
+		fmt.Println("adding did: ", d)
+		b.relevantDids[d] = true
+	}
+
+	return nil
+}
+
+type SyncInfo struct {
+	Repo          uint `gorm:"index"`
+	FollowsSynced bool
+	Rev           string
+}
+
+func (b *PostgresBackend) ensureFollowsScraped(ctx context.Context, user string) error {
+	r, err := b.GetOrCreateRepo(ctx, user)
+	if err != nil {
+		return err
+	}
+
+	var si SyncInfo
+	if err := b.db.Find(&si, "repo = ?", r.ID).Error; err != nil {
+		return err
+	}
+
+	// not found
+	if si.Repo == 0 {
+		if err := b.db.Create(&SyncInfo{
+			Repo: r.ID,
+		}).Error; err != nil {
+			return err
+		}
+	}
+
+	if si.FollowsSynced {
+		return nil
+	}
+
+	var follows []Follow
+	var cursor string
+	for {
+		resp, err := atproto.RepoListRecords(ctx, b.client, "app.bsky.graph.follow", cursor, 100, b.mydid, false)
+		if err != nil {
+			return err
+		}
+
+		for _, rec := range resp.Records {
+			if fol, ok := rec.Value.Val.(*bsky.GraphFollow); ok {
+				fr, err := b.GetOrCreateRepo(ctx, fol.Subject)
+				if err != nil {
+					return err
+				}
+
+				puri, err := syntax.ParseATURI(rec.Uri)
+				if err != nil {
+					return err
+				}
+
+				follows = append(follows, Follow{
+					Created: time.Now(),
+					Indexed: time.Now(),
+					Rkey:    puri.RecordKey().String(),
+					Author:  r.ID,
+					Subject: fr.ID,
+				})
+			}
+		}
+
+		if resp.Cursor == nil || len(resp.Records) == 0 {
+			break
+		}
+		cursor = *resp.Cursor
+	}
+
+	if err := b.db.Clauses(clause.OnConflict{DoNothing: true}).CreateInBatches(follows, 200).Error; err != nil {
+		return err
+	}
+
+	if err := b.db.Model(SyncInfo{}).Where("repo = ?", r.ID).Update("follows_synced", true).Error; err != nil {
+		return err
+	}
+
+	fmt.Println("Got follows: ", len(follows))
+
+	return nil
+}

events.go

···
-package main
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"log/slog"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/bluesky-social/indigo/api/atproto"
-	"github.com/bluesky-social/indigo/api/bsky"
-	"github.com/bluesky-social/indigo/atproto/syntax"
-	"github.com/bluesky-social/indigo/repo"
-	lru "github.com/hashicorp/golang-lru/v2"
-	"github.com/ipfs/go-cid"
-	"github.com/jackc/pgx/v5/pgconn"
-	"github.com/jackc/pgx/v5/pgxpool"
-	"gorm.io/gorm"
-
-	. "github.com/whyrusleeping/konbini/models"
-)
-
-type PostgresBackend struct {
-	db  *gorm.DB
-	pgx *pgxpool.Pool
-	s   *Server
-
-	relevantDids map[string]bool
-	rdLk         sync.Mutex
-
-	revCache *lru.TwoQueueCache[uint, string]
-
-	repoCache *lru.TwoQueueCache[string, *Repo]
-	reposLk   sync.Mutex
-
-	postInfoCache *lru.TwoQueueCache[string, cachedPostInfo]
-}
-
-func (b *PostgresBackend) HandleEvent(ctx context.Context, evt *atproto.SyncSubscribeRepos_Commit) error {
-	r, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(evt.Blocks))
-	if err != nil {
-		return fmt.Errorf("failed to read event repo: %w", err)
-	}
-
-	for _, op := range evt.Ops {
-		switch op.Action {
-		case "create":
-			c, rec, err := r.GetRecordBytes(ctx, op.Path)
-			if err != nil {
-				return err
-			}
-			if err := b.HandleCreate(ctx, evt.Repo, evt.Rev, op.Path, rec, &c); err != nil {
-				return fmt.Errorf("create record failed: %w", err)
-			}
-		case "update":
-			c, rec, err := r.GetRecordBytes(ctx, op.Path)
-			if err != nil {
-				return err
-			}
-			if err := b.HandleUpdate(ctx, evt.Repo, evt.Rev, op.Path, rec, &c); err != nil {
-				return fmt.Errorf("update record failed: %w", err)
-			}
-		case "delete":
-			if err := b.HandleDelete(ctx, evt.Repo, evt.Rev, op.Path); err != nil {
-				return fmt.Errorf("delete record failed: %w", err)
-			}
-		}
-	}
-
-	// TODO: sync with the Since field to make sure we don't miss events we care about
-	/*
-		if err := bf.Store.UpdateRev(ctx, evt.Repo, evt.Rev); err != nil {
-			return fmt.Errorf("failed to update rev: %w", err)
-		}
-	*/
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreate(ctx context.Context, repo string, rev string, path string, rec *[]byte, cid *cid.Cid) error {
-	start := time.Now()
-
-	rr, err := b.getOrCreateRepo(ctx, repo)
-	if err != nil {
-		return fmt.Errorf("get user failed: %w", err)
-	}
-
-	lrev, err := b.revForRepo(rr)
-	if err != nil {
-		return err
-	}
-	if lrev != "" {
-		if rev < lrev {
-			slog.Info("skipping old rev create", "did", rr.Did, "rev", rev, "oldrev", lrev, "path", path)
-			return nil
-		}
-	}
-
-	parts := strings.Split(path, "/")
-	if len(parts) != 2 {
-		return fmt.Errorf("invalid path in HandleCreate: %q", path)
-	}
-	col := parts[0]
-	rkey := parts[1]
-
-	defer func() {
-		handleOpHist.WithLabelValues("create", col).Observe(float64(time.Since(start).Milliseconds()))
-	}()
-
-	if rkey == "" {
-		fmt.Printf("messed up path: %q\n", rkey)
-	}
-
-	switch col {
-	case "app.bsky.feed.post":
-		if err := b.HandleCreatePost(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.feed.like":
-		if err := b.HandleCreateLike(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.feed.repost":
-		if err := b.HandleCreateRepost(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.graph.follow":
-		if err := b.HandleCreateFollow(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.graph.block":
-		if err := b.HandleCreateBlock(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.graph.list":
-		if err := b.HandleCreateList(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.graph.listitem":
-		if err := b.HandleCreateListitem(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.graph.listblock":
-		if err := b.HandleCreateListblock(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.actor.profile":
-		if err := b.HandleCreateProfile(ctx, rr, rkey, rev, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.feed.generator":
-		if err := b.HandleCreateFeedGenerator(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.feed.threadgate":
-		if err := b.HandleCreateThreadgate(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "chat.bsky.actor.declaration":
-		if err := b.HandleCreateChatDeclaration(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.feed.postgate":
-		if err := b.HandleCreatePostGate(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	case "app.bsky.graph.starterpack":
-		if err := b.HandleCreateStarterPack(ctx, rr, rkey, *rec, *cid); err != nil {
-			return err
-		}
-	default:
-		slog.Debug("unrecognized record type", "repo", repo, "path", path, "rev", rev)
-	}
-
-	b.revCache.Add(rr.ID, rev)
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreatePost(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	exists, err := b.checkPostExists(ctx, repo, rkey)
-	if err != nil {
-		return err
-	}
-
-	// still technically a race condition if two creates for the same post happen concurrently... probably fine
-	if exists {
-		return nil
-	}
-
-	var rec bsky.FeedPost
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		uri := "at://" + repo.Did + "/app.bsky.feed.post/" + rkey
-		slog.Warn("skipping post with malformed data", "uri", uri, "error", err)
-		return nil // Skip this post rather than failing the entire event
-	}
-
-	reldids := []string{repo.Did}
-	// care about a post if its in a thread of a user we are interested in
-	if rec.Reply != nil && rec.Reply.Parent != nil && rec.Reply.Root != nil {
-		reldids = append(reldids, rec.Reply.Parent.Uri, rec.Reply.Root.Uri)
-	}
-	// TODO: maybe also care if its mentioning a user we care about or quoting a user we care about?
-	if !b.anyRelevantIdents(reldids...) {
-		return nil
-	}
-
-	uri := "at://" + repo.Did + "/app.bsky.feed.post/" + rkey
-	slog.Warn("adding post", "uri", uri)
-
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	p := Post{
-		Created: created.Time(),
-		Indexed: time.Now(),
-		Author:  repo.ID,
-		Rkey:    rkey,
-		Raw:     recb,
-		Cid:     cc.String(),
-	}
-
-	if rec.Reply != nil && rec.Reply.Parent != nil {
-		if rec.Reply.Root == nil {
-			return fmt.Errorf("post reply had nil root")
-		}
-
-		pinfo, err := b.postInfoForUri(ctx, rec.Reply.Parent.Uri)
-		if err != nil {
-			return fmt.Errorf("getting reply parent: %w", err)
-		}
-
-		p.ReplyTo = pinfo.ID
-		p.ReplyToUsr = pinfo.Author
-
-		thread, err := b.postIDForUri(ctx, rec.Reply.Root.Uri)
-		if err != nil {
-			return fmt.Errorf("getting thread root: %w", err)
-		}
-
-		p.InThread = thread
-
-		if p.ReplyToUsr == b.s.myrepo.ID {
-			if err := b.s.AddNotification(ctx, b.s.myrepo.ID, p.Author, uri, cc, NotifKindReply); err != nil {
-				slog.Warn("failed to create notification", "uri", uri, "error", err)
-			}
-		}
-	}
-
-	if rec.Embed != nil {
-		var rpref string
-		if rec.Embed.EmbedRecord != nil && rec.Embed.EmbedRecord.Record != nil {
-			rpref = rec.Embed.EmbedRecord.Record.Uri
-		}
-		if rec.Embed.EmbedRecordWithMedia != nil &&
-			rec.Embed.EmbedRecordWithMedia.Record != nil &&
-			rec.Embed.EmbedRecordWithMedia.Record.Record != nil {
-			rpref = rec.Embed.EmbedRecordWithMedia.Record.Record.Uri
-		}
-
-		if rpref != "" && strings.Contains(rpref, "app.bsky.feed.post") {
-			rp, err := b.postIDForUri(ctx, rpref)
-			if err != nil {
-				return fmt.Errorf("getting quote subject: %w", err)
-			}
-
-			p.Reposting = rp
-		}
-	}
-
-	if err := b.doPostCreate(ctx, &p); err != nil {
-		return err
-	}
-
-	// Check for mentions and create notifications
-	if rec.Facets != nil {
-		for _, facet := range rec.Facets {
-			for _, feature := range facet.Features {
-				if feature.RichtextFacet_Mention != nil {
-					mentionDid := feature.RichtextFacet_Mention.Did
-					// This is a mention
-					mentionedRepo, err := b.getOrCreateRepo(ctx, mentionDid)
-					if err != nil {
-						slog.Warn("failed to get repo for mention", "did", mentionDid, "error", err)
-						continue
-					}
-
-					// Create notification if the mentioned user is the current user
-					if mentionedRepo.ID == b.s.myrepo.ID {
-						if err := b.s.AddNotification(ctx, b.s.myrepo.ID, p.Author, uri, cc, NotifKindMention); err != nil {
-							slog.Warn("failed to create mention notification", "uri", uri, "error", err)
-						}
-					}
-				}
-			}
-		}
-	}
-
-	b.postInfoCache.Add(uri, cachedPostInfo{
-		ID:     p.ID,
-		Author: p.Author,
-	})
-
-	return nil
-}
-
-func (b *PostgresBackend) doPostCreate(ctx context.Context, p *Post) error {
-	/*
-		if err := b.db.Clauses(clause.OnConflict{
-			Columns:   []clause.Column{{Name: "author"}, {Name: "rkey"}},
-			DoUpdates: clause.AssignmentColumns([]string{"cid", "not_found", "raw", "created", "indexed"}),
-		}).Create(p).Error; err != nil {
-			return err
-		}
-	*/
-
-	query := `
-INSERT INTO posts (author, rkey, cid, not_found, raw, created, indexed, reposting, reply_to, reply_to_usr, in_thread) 
-VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
-ON CONFLICT (author, rkey) 
-DO UPDATE SET 
-    cid = $3,
-    not_found = $4,
-    raw = $5,
-    created = $6,
-    indexed = $7,
-    reposting = $8,
-    reply_to = $9,
-    reply_to_usr = $10,
-    in_thread = $11
-RETURNING id
-`
-
-	// Execute the query with parameters from the Post struct
-	if err := b.pgx.QueryRow(
-		ctx,
-		query,
-		p.Author,
-		p.Rkey,
-		p.Cid,
-		p.NotFound,
-		p.Raw,
-		p.Created,
-		p.Indexed,
-		p.Reposting,
-		p.ReplyTo,
-		p.ReplyToUsr,
-		p.InThread,
-	).Scan(&p.ID); err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateLike(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	var rec bsky.FeedLike
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		return err
-	}
-
-	if !b.anyRelevantIdents(repo.Did, rec.Subject.Uri) {
-		return nil
-	}
-
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	pinfo, err := b.postInfoForUri(ctx, rec.Subject.Uri)
-	if err != nil {
-		return fmt.Errorf("getting like subject: %w", err)
-	}
-
-	if _, err := b.pgx.Exec(ctx, `INSERT INTO "likes" ("created","indexed","author","rkey","subject","cid") VALUES ($1, $2, $3, $4, $5, $6)`, created.Time(), time.Now(), repo.ID, rkey, pinfo.ID, cc.String()); err != nil {
-		pgErr, ok := err.(*pgconn.PgError)
-		if ok && pgErr.Code == "23505" {
-			return nil
-		}
-		return err
-	}
-
-	// Create notification if the liked post belongs to the current user
-	if pinfo.Author == b.s.myrepo.ID {
-		uri := fmt.Sprintf("at://%s/app.bsky.feed.like/%s", repo.Did, rkey)
-		if err := b.s.AddNotification(ctx, b.s.myrepo.ID, repo.ID, uri, cc, NotifKindLike); err != nil {
-			slog.Warn("failed to create like notification", "uri", uri, "error", err)
-		}
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateRepost(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	var rec bsky.FeedRepost
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		return err
-	}
-
-	if !b.anyRelevantIdents(repo.Did, rec.Subject.Uri) {
-		return nil
-	}
-
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	pinfo, err := b.postInfoForUri(ctx, rec.Subject.Uri)
-	if err != nil {
-		return fmt.Errorf("getting repost subject: %w", err)
-	}
-
-	if _, err := b.pgx.Exec(ctx, `INSERT INTO "reposts" ("created","indexed","author","rkey","subject") VALUES ($1, $2, $3, $4, $5)`, created.Time(), time.Now(), repo.ID, rkey, pinfo.ID); err != nil {
-		pgErr, ok := err.(*pgconn.PgError)
-		if ok && pgErr.Code == "23505" {
-			return nil
-		}
-		return err
-	}
-
-	// Create notification if the reposted post belongs to the current user
-	if pinfo.Author == b.s.myrepo.ID {
-		uri := fmt.Sprintf("at://%s/app.bsky.feed.repost/%s", repo.Did, rkey)
-		if err := b.s.AddNotification(ctx, b.s.myrepo.ID, repo.ID, uri, cc, NotifKindRepost); err != nil {
-			slog.Warn("failed to create repost notification", "uri", uri, "error", err)
-		}
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateFollow(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	var rec bsky.GraphFollow
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		return err
-	}
-
-	if !b.anyRelevantIdents(repo.Did, rec.Subject) {
-		return nil
-	}
-
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	subj, err := b.getOrCreateRepo(ctx, rec.Subject)
-	if err != nil {
-		return err
-	}
-
-	if _, err := b.pgx.Exec(ctx, "INSERT INTO follows (created, indexed, author, rkey, subject) VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING", created.Time(), time.Now(), repo.ID, rkey, subj.ID); err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateBlock(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	var rec bsky.GraphBlock
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		return err
-	}
-
-	if !b.anyRelevantIdents(repo.Did, rec.Subject) {
-		return nil
-	}
-
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	subj, err := b.getOrCreateRepo(ctx, rec.Subject)
-	if err != nil {
-		return err
-	}
-
-	if err := b.db.Create(&Block{
-		Created: created.Time(),
-		Indexed: time.Now(),
-		Author:  repo.ID,
-		Rkey:    rkey,
-		Subject: subj.ID,
-	}).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateList(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	var rec bsky.GraphList
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		return err
-	}
-
-	if !b.anyRelevantIdents(repo.Did) {
-		return nil
-	}
-
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	if err := b.db.Create(&List{
-		Created: created.Time(),
-		Indexed: time.Now(),
-		Author:  repo.ID,
-		Rkey:    rkey,
-		Raw:     recb,
-	}).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateListitem(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	var rec bsky.GraphListitem
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		return err
-	}
-	if !b.anyRelevantIdents(repo.Did) {
-		return nil
-	}
-
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	subj, err := b.getOrCreateRepo(ctx, rec.Subject)
-	if err != nil {
-		return err
-	}
-
-	list, err := b.getOrCreateList(ctx, rec.List)
-	if err != nil {
-		return err
-	}
-
-	if err := b.db.Create(&ListItem{
-		Created: created.Time(),
-		Indexed: time.Now(),
-		Author:  repo.ID,
-		Rkey:    rkey,
-		Subject: subj.ID,
-		List:    list.ID,
-	}).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateListblock(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	var rec bsky.GraphListblock
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		return err
-	}
-
-	if !b.anyRelevantIdents(repo.Did, rec.Subject) {
-		return nil
-	}
-
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	list, err := b.getOrCreateList(ctx, rec.Subject)
-	if err != nil {
-		return err
-	}
-
-	if err := b.db.Create(&ListBlock{
-		Created: created.Time(),
-		Indexed: time.Now(),
-		Author:  repo.ID,
-		Rkey:    rkey,
-		List:    list.ID,
-	}).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateProfile(ctx context.Context, repo *Repo, rkey, rev string, recb []byte, cc cid.Cid) error {
-	if !b.anyRelevantIdents(repo.Did) {
-		return nil
-	}
-
-	if err := b.db.Create(&Profile{
-		//Created: created.Time(),
-		Indexed: time.Now(),
-		Repo:    repo.ID,
-		Raw:     recb,
-		Rev:     rev,
-	}).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleUpdateProfile(ctx context.Context, repo *Repo, rkey, rev string, recb []byte, cc cid.Cid) error {
-	if !b.anyRelevantIdents(repo.Did) {
-		return nil
-	}
-
-	if err := b.db.Create(&Profile{
-		Indexed: time.Now(),
-		Repo:    repo.ID,
-		Raw:     recb,
-		Rev:     rev,
-	}).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateFeedGenerator(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	if !b.anyRelevantIdents(repo.Did) {
-		return nil
-	}
-
-	var rec bsky.FeedGenerator
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		return err
-	}
-
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	if err := b.db.Create(&FeedGenerator{
-		Created: created.Time(),
-		Indexed: time.Now(),
-		Author:  repo.ID,
-		Rkey:    rkey,
-		Did:     rec.Did,
-		Raw:     recb,
-	}).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateThreadgate(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	if !b.anyRelevantIdents(repo.Did) {
-		return nil
-	}
-	var rec bsky.FeedThreadgate
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		return err
-	}
-
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	pid, err := b.postIDForUri(ctx, rec.Post)
-	if err != nil {
-		return err
-	}
-
-	if err := b.db.Create(&ThreadGate{
-		Created: created.Time(),
-		Indexed: time.Now(),
-		Author:  repo.ID,
-		Rkey:    rkey,
-		Post:    pid,
-	}).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateChatDeclaration(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	// TODO: maybe track these?
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreatePostGate(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	if !b.anyRelevantIdents(repo.Did) {
-		return nil
-	}
-	var rec bsky.FeedPostgate
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		return err
-	}
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	refPost, err := b.postInfoForUri(ctx, rec.Post)
-	if err != nil {
-		return err
-	}
-
-	if err := b.db.Create(&PostGate{
-		Created: created.Time(),
-		Indexed: time.Now(),
-		Author:  repo.ID,
-		Rkey:    rkey,
-		Subject: refPost.ID,
-		Raw:     recb,
-	}).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleCreateStarterPack(ctx context.Context, repo *Repo, rkey string, recb []byte, cc cid.Cid) error {
-	if !b.anyRelevantIdents(repo.Did) {
-		return nil
-	}
-	var rec bsky.GraphStarterpack
-	if err := rec.UnmarshalCBOR(bytes.NewReader(recb)); err != nil {
-		return err
-	}
-	created, err := syntax.ParseDatetimeLenient(rec.CreatedAt)
-	if err != nil {
-		return fmt.Errorf("invalid timestamp: %w", err)
-	}
-
-	list, err := b.getOrCreateList(ctx, rec.List)
-	if err != nil {
-		return err
-	}
-
-	if err := b.db.Create(&StarterPack{
-		Created: created.Time(),
-		Indexed: time.Now(),
-		Author:  repo.ID,
-		Rkey:    rkey,
-		Raw:     recb,
-		List:    list.ID,
-	}).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleUpdate(ctx context.Context, repo string, rev string, path string, rec *[]byte, cid *cid.Cid) error {
-	start := time.Now()
-
-	rr, err := b.getOrCreateRepo(ctx, repo)
-	if err != nil {
-		return fmt.Errorf("get user failed: %w", err)
-	}
-
-	lrev, err := b.revForRepo(rr)
-	if err != nil {
-		return err
-	}
-	if lrev != "" {
-		if rev < lrev {
-			//slog.Info("skipping old rev create", "did", rr.Did, "rev", rev, "oldrev", lrev, "path", path)
-			return nil
-		}
-	}
-
-	parts := strings.Split(path, "/")
-	if len(parts) != 2 {
-		return fmt.Errorf("invalid path in HandleCreate: %q", path)
-	}
-	col := parts[0]
-	rkey := parts[1]
-
-	defer func() {
-		handleOpHist.WithLabelValues("update", col).Observe(float64(time.Since(start).Milliseconds()))
-	}()
-
-	if rkey == "" {
-		fmt.Printf("messed up path: %q\n", rkey)
-	}
-
-	switch col {
-	/*
-		case "app.bsky.feed.post":
-			if err := s.HandleCreatePost(ctx, rr, rkey, *rec, *cid); err != nil {
-				return err
-			}
-		case "app.bsky.feed.like":
-			if err := s.HandleCreateLike(ctx, rr, rkey, *rec, *cid); err != nil {
-				return err
-			}
-		case "app.bsky.feed.repost":
-			if err := s.HandleCreateRepost(ctx, rr, rkey, *rec, *cid); err != nil {
-				return err
-			}
-		case "app.bsky.graph.follow":
-			if err := s.HandleCreateFollow(ctx, rr, rkey, *rec, *cid); err != nil {
-				return err
-			}
-		case "app.bsky.graph.block":
-			if err := s.HandleCreateBlock(ctx, rr, rkey, *rec, *cid); err != nil {
-				return err
-			}
-		case "app.bsky.graph.list":
-			if err := s.HandleCreateList(ctx, rr, rkey, *rec, *cid); err != nil {
-				return err
-			}
-		case "app.bsky.graph.listitem":
-			if err := s.HandleCreateListitem(ctx, rr, rkey, *rec, *cid); err != nil {
-				return err
-			}
-		case "app.bsky.graph.listblock":
-			if err := s.HandleCreateListblock(ctx, rr, rkey, *rec, *cid); err != nil {
-				return err
-			}
-	*/
-	case "app.bsky.actor.profile":
-		if err := b.HandleUpdateProfile(ctx, rr, rkey, rev, *rec, *cid); err != nil {
-			return err
-		}
-		/*
-			case "app.bsky.feed.generator":
-				if err := s.HandleCreateFeedGenerator(ctx, rr, rkey, *rec, *cid); err != nil {
-					return err
-				}
-			case "app.bsky.feed.threadgate":
-				if err := s.HandleCreateThreadgate(ctx, rr, rkey, *rec, *cid); err != nil {
-					return err
-				}
-			case "chat.bsky.actor.declaration":
-				if err := s.HandleCreateChatDeclaration(ctx, rr, rkey, *rec, *cid); err != nil {
-					return err
-				}
-		*/
-	default:
-		slog.Debug("unrecognized record type in update", "repo", repo, "path", path, "rev", rev)
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleDelete(ctx context.Context, repo string, rev string, path string) error {
-	start := time.Now()
-
-	rr, err := b.getOrCreateRepo(ctx, repo)
-	if err != nil {
-		return fmt.Errorf("get user failed: %w", err)
-	}
-
-	lrev, ok := b.revCache.Get(rr.ID)
-	if ok {
-		if rev < lrev {
-			//slog.Info("skipping old rev delete", "did", rr.Did, "rev", rev, "oldrev", lrev)
-			return nil
-		}
-	}
-
-	parts := strings.Split(path, "/")
-	if len(parts) != 2 {
-		return fmt.Errorf("invalid path in HandleDelete: %q", path)
-	}
-	col := parts[0]
-	rkey := parts[1]
-
-	defer func() {
-		handleOpHist.WithLabelValues("create", col).Observe(float64(time.Since(start).Milliseconds()))
-	}()
-
-	switch col {
-	case "app.bsky.feed.post":
-		if err := b.HandleDeletePost(ctx, rr, rkey); err != nil {
-			return err
-		}
-	case "app.bsky.feed.like":
-		if err := b.HandleDeleteLike(ctx, rr, rkey); err != nil {
-			return err
-		}
-	case "app.bsky.feed.repost":
-		if err := b.HandleDeleteRepost(ctx, rr, rkey); err != nil {
-			return err
-		}
-	case "app.bsky.graph.follow":
-		if err := b.HandleDeleteFollow(ctx, rr, rkey); err != nil {
-			return err
-		}
-	case "app.bsky.graph.block":
-		if err := b.HandleDeleteBlock(ctx, rr, rkey); err != nil {
-			return err
-		}
-	case "app.bsky.graph.list":
-		if err := b.HandleDeleteList(ctx, rr, rkey); err != nil {
-			return err
-		}
-	case "app.bsky.graph.listitem":
-		if err := b.HandleDeleteListitem(ctx, rr, rkey); err != nil {
-			return err
-		}
-	case "app.bsky.graph.listblock":
-		if err := b.HandleDeleteListblock(ctx, rr, rkey); err != nil {
-			return err
-		}
-	case "app.bsky.actor.profile":
-		if err := b.HandleDeleteProfile(ctx, rr, rkey); err != nil {
-			return err
-		}
-	case "app.bsky.feed.generator":
-		if err := b.HandleDeleteFeedGenerator(ctx, rr, rkey); err != nil {
-			return err
-		}
-	case "app.bsky.feed.threadgate":
-		if err := b.HandleDeleteThreadgate(ctx, rr, rkey); err != nil {
-			return err
-		}
-	default:
-		slog.Warn("delete unrecognized record type", "repo", repo, "path", path, "rev", rev)
-	}
-
-	b.revCache.Add(rr.ID, rev)
-	return nil
-}
-
-func (b *PostgresBackend) HandleDeletePost(ctx context.Context, repo *Repo, rkey string) error {
-	var p Post
-	if err := b.db.Find(&p, "author = ? AND rkey = ?", repo.ID, rkey).Error; err != nil {
-		return err
-	}
-
-	if p.ID == 0 {
-		//slog.Warn("delete of unknown post record", "repo", repo.Did, "rkey", rkey)
-		return nil
-	}
-
-	if err := b.db.Delete(&Post{}, p.ID).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleDeleteLike(ctx context.Context, repo *Repo, rkey string) error {
-	var like Like
-	if err := b.db.Find(&like, "author = ? AND rkey = ?", repo.ID, rkey).Error; err != nil {
-		return err
-	}
-
-	if like.ID == 0 {
-		//slog.Warn("delete of missing like", "repo", repo.Did, "rkey", rkey)
-		return nil
-	}
-
-	if err := b.db.Exec("DELETE FROM likes WHERE id = ?", like.ID).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleDeleteRepost(ctx context.Context, repo *Repo, rkey string) error {
-	var repost Repost
-	if err := b.db.Find(&repost, "author = ? AND rkey = ?", repo.ID, rkey).Error; err != nil {
-		return err
-	}
-
-	if repost.ID == 0 {
-		//return fmt.Errorf("delete of missing repost: %s %s", repo.Did, rkey)
-		return nil
-	}
-
-	if err := b.db.Exec("DELETE FROM reposts WHERE id = ?", repost.ID).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleDeleteFollow(ctx context.Context, repo *Repo, rkey string) error {
-	var follow Follow
-	if err := b.db.Find(&follow, "author = ? AND rkey = ?", repo.ID, rkey).Error; err != nil {
-		return err
-	}
-
-	if follow.ID == 0 {
-		//slog.Warn("delete of missing follow", "repo", repo.Did, "rkey", rkey)
-		return nil
-	}
-
-	if err := b.db.Exec("DELETE FROM follows WHERE id = ?", follow.ID).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleDeleteBlock(ctx context.Context, repo *Repo, rkey string) error {
-	var block Block
-	if err := b.db.Find(&block, "author = ? AND rkey = ?", repo.ID, rkey).Error; err != nil {
-		return err
-	}
-
-	if block.ID == 0 {
-		//slog.Warn("delete of missing block", "repo", repo.Did, "rkey", rkey)
-		return nil
-	}
-
-	if err := b.db.Exec("DELETE FROM blocks WHERE id = ?", block.ID).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleDeleteList(ctx context.Context, repo *Repo, rkey string) error {
-	var list List
-	if err := b.db.Find(&list, "author = ? AND rkey = ?", repo.ID, rkey).Error; err != nil {
-		return err
-	}
-
-	if list.ID == 0 {
-		return nil
-		//return fmt.Errorf("delete of missing list: %s %s", repo.Did, rkey)
-	}
-
-	if err := b.db.Exec("DELETE FROM lists WHERE id = ?", list.ID).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleDeleteListitem(ctx context.Context, repo *Repo, rkey string) error {
-	var item ListItem
-	if err := b.db.Find(&item, "author = ? AND rkey = ?", repo.ID, rkey).Error; err != nil {
-		return err
-	}
-
-	if item.ID == 0 {
-		return nil
-		//return fmt.Errorf("delete of missing listitem: %s %s", repo.Did, rkey)
-	}
-
-	if err := b.db.Exec("DELETE FROM list_items WHERE id = ?", item.ID).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleDeleteListblock(ctx context.Context, repo *Repo, rkey string) error {
-	var block ListBlock
-	if err := b.db.Find(&block, "author = ? AND rkey = ?", repo.ID, rkey).Error; err != nil {
-		return err
-	}
-
-	if block.ID == 0 {
-		return nil
-		//return fmt.Errorf("delete of missing listblock: %s %s", repo.Did, rkey)
-	}
-
-	if err := b.db.Exec("DELETE FROM list_blocks WHERE id = ?", block.ID).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleDeleteFeedGenerator(ctx context.Context, repo *Repo, rkey string) error {
-	var feedgen FeedGenerator
-	if err := b.db.Find(&feedgen, "author = ? AND rkey = ?", repo.ID, rkey).Error; err != nil {
-		return err
-	}
-
-	if feedgen.ID == 0 {
-		return nil
-		//return fmt.Errorf("delete of missing feedgen: %s %s", repo.Did, rkey)
-	}
-
-	if err := b.db.Exec("DELETE FROM feed_generators WHERE id = ?", feedgen.ID).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleDeleteThreadgate(ctx context.Context, repo *Repo, rkey string) error {
-	var threadgate ThreadGate
-	if err := b.db.Find(&threadgate, "author = ? AND rkey = ?", repo.ID, rkey).Error; err != nil {
-		return err
-	}
-
-	if threadgate.ID == 0 {
-		return nil
-		//return fmt.Errorf("delete of missing threadgate: %s %s", repo.Did, rkey)
-	}
-
-	if err := b.db.Exec("DELETE FROM thread_gates WHERE id = ?", threadgate.ID).Error; err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (b *PostgresBackend) HandleDeleteProfile(ctx context.Context, repo *Repo, rkey string) error {
-	var profile Profile
-	if err := b.db.Find(&profile, "repo = ?", repo.ID).Error; err != nil {
-		return err
-	}
-
-	if profile.ID == 0 {
-		return nil
-	}
-
-	if err := b.db.Exec("DELETE FROM profiles WHERE id = ?", profile.ID).Error; err != nil {
-		return err
-	}
-
-	return nil
-}

handlers.go

···
 	"github.com/labstack/gommon/log"
 	"github.com/whyrusleeping/market/models"
 
+	"github.com/whyrusleeping/konbini/backend"
 	. "github.com/whyrusleeping/konbini/models"
 )
 
···
 
 func (s *Server) handleGetRelevantDids(e echo.Context) error {
 	return e.JSON(200, map[string]any{
-		"dids": s.backend.relevantDids,
+		"dids": s.backend.GetRelevantDids(),
 	})
 }
 
···
 
 	postUri := fmt.Sprintf("at://%s/app.bsky.feed.post/%s", did, rkey)
 
-	p, err := s.backend.getPostByUri(ctx, postUri, "*")
+	p, err := s.backend.GetPostByUri(ctx, postUri, "*")
 	if err != nil {
 		return err
 	}
···
 		return err
 	}
 
-	r, err := s.backend.getOrCreateRepo(ctx, accdid)
+	r, err := s.backend.GetOrCreateRepo(ctx, accdid)
 	if err != nil {
 		return err
 	}
 
 	var profile models.Profile
-	if err := s.backend.db.Find(&profile, "repo = ?", r.ID).Error; err != nil {
+	if err := s.db.Find(&profile, "repo = ?", r.ID).Error; err != nil {
 		return err
 	}
 
···
 		return err
 	}
 
-	r, err := s.backend.getOrCreateRepo(ctx, accdid)
+	r, err := s.backend.GetOrCreateRepo(ctx, accdid)
 	if err != nil {
 		return err
 	}
···
 	}
 
 	var dbposts []models.Post
-	if err := s.backend.db.Raw("SELECT * FROM posts WHERE author = ? AND created < ? ORDER BY created DESC LIMIT ?", r.ID, tcursor, limit).Scan(&dbposts).Error; err != nil {
+	if err := s.db.Raw("SELECT * FROM posts WHERE author = ? AND created < ? ORDER BY created DESC LIMIT ?", r.ID, tcursor, limit).Scan(&dbposts).Error; err != nil {
 		return err
 	}
 
···
 func (s *Server) handleGetFollowingFeed(e echo.Context) error {
 	ctx := e.Request().Context()
 
-	myr, err := s.backend.getOrCreateRepo(ctx, s.mydid)
+	myr, err := s.backend.GetOrCreateRepo(ctx, s.mydid)
 	if err != nil {
 		return err
 	}
···
 		tcursor = t
 	}
 	var dbposts []models.Post
-	if err := s.backend.db.Raw("select * from posts where reply_to = 0 AND author IN (select subject from follows where author = ?) AND created < ? order by created DESC limit ?", myr.ID, tcursor, limit).Scan(&dbposts).Error; err != nil {
+	if err := s.db.Raw("select * from posts where reply_to = 0 AND author IN (select subject from follows where author = ?) AND created < ? order by created DESC limit ?", myr.ID, tcursor, limit).Scan(&dbposts).Error; err != nil {
 		return err
 	}
 
···
 
 func (s *Server) getAuthorInfo(ctx context.Context, r *models.Repo) (*authorInfo, error) {
 	var profile models.Profile
-	if err := s.backend.db.Find(&profile, "repo = ?", r.ID).Error; err != nil {
+	if err := s.db.Find(&profile, "repo = ?", r.ID).Error; err != nil {
 		return nil, err
 	}
 
···
 
 	go func() {
 		defer wg.Done()
-		if err := s.backend.db.Raw("SELECT count(*) FROM likes WHERE subject = ?", pid).Scan(&pc.Likes).Error; err != nil {
+		if err := s.db.Raw("SELECT count(*) FROM likes WHERE subject = ?", pid).Scan(&pc.Likes).Error; err != nil {
 			slog.Error("failed to get likes count", "post", pid, "error", err)
 		}
 	}()
 
 	go func() {
 		defer wg.Done()
-		if err := s.backend.db.Raw("SELECT count(*) FROM reposts WHERE subject = ?", pid).Scan(&pc.Reposts).Error; err != nil {
+		if err := s.db.Raw("SELECT count(*) FROM reposts WHERE subject = ?", pid).Scan(&pc.Reposts).Error; err != nil {
 			slog.Error("failed to get reposts count", "post", pid, "error", err)
 		}
 	}()
 
 	go func() {
 		defer wg.Done()
-		if err := s.backend.db.Raw("SELECT count(*) FROM posts WHERE reply_to = ?", pid).Scan(&pc.Replies).Error; err != nil {
+		if err := s.db.Raw("SELECT count(*) FROM posts WHERE reply_to = ?", pid).Scan(&pc.Replies).Error; err != nil {
 			slog.Error("failed to get replies count", "post", pid, "error", err)
 		}
 	}()
···
 		go func(ix int) {
 			defer wg.Done()
 			p := dbposts[ix]
-			r, err := s.backend.getRepoByID(ctx, p.Author)
+			r, err := s.backend.GetRepoByID(ctx, p.Author)
 			if err != nil {
 				fmt.Println("failed to get repo: ", err)
 				posts[ix] = postResponse{
···
 
 func (s *Server) checkViewerLike(ctx context.Context, pid uint) *viewerLike {
 	var like Like
-	if err := s.backend.db.Raw("SELECT * FROM likes WHERE subject = ? AND author = ?", pid, s.myrepo.ID).Scan(&like).Error; err != nil {
+	if err := s.db.Raw("SELECT * FROM likes WHERE subject = ? AND author = ?", pid, s.myrepo.ID).Scan(&like).Error; err != nil {
 		slog.Error("failed to lookup like", "error", err)
 		return nil
 	}
···
 	quotedURI := embedRecord.Record.Uri
 	quotedCid := embedRecord.Record.Cid
 
-	quotedPost, err := s.backend.getPostByUri(ctx, quotedURI, "*")
+	quotedPost, err := s.backend.GetPostByUri(ctx, quotedURI, "*")
 	if err != nil {
 		slog.Warn("failed to get quoted post", "uri", quotedURI, "error", err)
 		s.addMissingPost(ctx, quotedURI)
···
 		return s.buildQuoteFallback(quotedURI, quotedCid)
 	}
 
-	quotedRepo, err := s.backend.getRepoByID(ctx, quotedPost.Author)
+	quotedRepo, err := s.backend.GetRepoByID(ctx, quotedPost.Author)
 	if err != nil {
 		slog.Warn("failed to get quoted post author", "error", err)
 		return s.buildQuoteFallback(quotedURI, quotedCid)
···
 
 	// Get the requested post to find the thread root
 	var requestedPost models.Post
-	if err := s.backend.db.Find(&requestedPost, "id = ?", postID).Error; err != nil {
+	if err := s.db.Find(&requestedPost, "id = ?", postID).Error; err != nil {
 		return err
 	}
 
···
 	// Get all posts in this thread
 	var dbposts []models.Post
 	query := "SELECT * FROM posts WHERE id = ? OR in_thread = ? ORDER BY created ASC"
-	if err := s.backend.db.Raw(query, rootPostID, rootPostID).Scan(&dbposts).Error; err != nil {
+	if err := s.db.Raw(query, rootPostID, rootPostID).Scan(&dbposts).Error; err != nil {
 		return err
 	}
 
 	// Build response for each post
 	posts := []postResponse{}
 	for _, p := range dbposts {
-		r, err := s.backend.getRepoByID(ctx, p.Author)
+		r, err := s.backend.GetRepoByID(ctx, p.Author)
 		if err != nil {
 			return err
 		}
···
 
 	// Get all likes for this post
 	var likes []models.Like
-	if err := s.backend.db.Find(&likes, "subject = ?", postID).Error; err != nil {
+	if err := s.db.Find(&likes, "subject = ?", postID).Error; err != nil {
 		return err
 	}
 
 	users := []engagementUser{}
 	for _, like := range likes {
-		r, err := s.backend.getRepoByID(ctx, like.Author)
+		r, err := s.backend.GetRepoByID(ctx, like.Author)
 		if err != nil {
 			slog.Error("failed to get repo for like author", "error", err)
 			continue
···
 
 		// Get profile if available
 		var profile models.Profile
-		s.backend.db.Find(&profile, "repo = ?", r.ID)
+		s.db.Find(&profile, "repo = ?", r.ID)
 
 		var prof *bsky.ActorProfile
 		if len(profile.Raw) > 0 {
···
 
 	// Get all reposts for this post
 	var reposts []models.Repost
-	if err := s.backend.db.Find(&reposts, "subject = ?", postID).Error; err != nil {
+	if err := s.db.Find(&reposts, "subject = ?", postID).Error; err != nil {
 		return err
 	}
 
 	users := []engagementUser{}
 	for _, repost := range reposts {
-		r, err := s.backend.getRepoByID(ctx, repost.Author)
+		r, err := s.backend.GetRepoByID(ctx, repost.Author)
 		if err != nil {
 			slog.Error("failed to get repo for repost author", "error", err)
 			continue
···
 
 		// Get profile if available
 		var profile models.Profile
-		s.backend.db.Find(&profile, "repo = ?", r.ID)
+		s.db.Find(&profile, "repo = ?", r.ID)
 
 		var prof *bsky.ActorProfile
 		if len(profile.Raw) > 0 {
···
 
 	// Get all replies to this post
 	var replies []models.Post
-	if err := s.backend.db.Find(&replies, "reply_to = ?", postID).Error; err != nil {
+	if err := s.db.Find(&replies, "reply_to = ?", postID).Error; err != nil {
 		return err
 	}
 
···
 		}
 		seen[reply.Author] = true
 
-		r, err := s.backend.getRepoByID(ctx, reply.Author)
+		r, err := s.backend.GetRepoByID(ctx, reply.Author)
 		if err != nil {
 			slog.Error("failed to get repo for reply author", "error", err)
 			continue
···
 
 		// Get profile if available
 		var profile models.Profile
-		s.backend.db.Find(&profile, "repo = ?", r.ID)
+		s.db.Find(&profile, "repo = ?", r.ID)
 
 		var prof *bsky.ActorProfile
 		if len(profile.Raw) > 0 {
···
 	query := `SELECT * FROM notifications WHERE "for" = ?`
 	if cursorID > 0 {
 		query += ` AND id < ?`
-		if err := s.backend.db.Raw(query+" ORDER BY created_at DESC LIMIT ?", s.myrepo.ID, cursorID, limit).Scan(&notifications).Error; err != nil {
+		if err := s.db.Raw(query+" ORDER BY created_at DESC LIMIT ?", s.myrepo.ID, cursorID, limit).Scan(&notifications).Error; err != nil {
 			return err
 		}
 	} else {
-		if err := s.backend.db.Raw(query+" ORDER BY created_at DESC LIMIT ?", s.myrepo.ID, limit).Scan(&notifications).Error; err != nil {
+		if err := s.db.Raw(query+" ORDER BY created_at DESC LIMIT ?", s.myrepo.ID, limit).Scan(&notifications).Error; err != nil {
 			return err
 		}
 	}
···
 	results := []notificationResponse{}
 	for _, notif := range notifications {
 		// Get author info
-		author, err := s.backend.getRepoByID(ctx, notif.Author)
+		author, err := s.backend.GetRepoByID(ctx, notif.Author)
 		if err != nil {
 			slog.Error("failed to get repo for notification author", "error", err)
 			continue
···
 		}
 
 		// Try to get source post preview for reply/mention notifications
-		if notif.Kind == NotifKindReply || notif.Kind == NotifKindMention {
+		if notif.Kind == backend.NotifKindReply || notif.Kind == backend.NotifKindMention {
 			// Parse URI to get post
-			p, err := s.backend.getPostByUri(ctx, notif.Source, "*")
+			p, err := s.backend.GetPostByUri(ctx, notif.Source, "*")
 			if err == nil && p.Raw != nil && len(p.Raw) > 0 {
 				var fp bsky.FeedPost
 				if err := fp.UnmarshalCBOR(bytes.NewReader(p.Raw)); err == nil {

hydration/hydrator.go

···
 type Hydrator struct {
 	db      *gorm.DB
 	dir     identity.Directory
-	backend backend.RecordTracker
+	backend *backend.PostgresBackend
 }
 
 // NewHydrator creates a new Hydrator
-func NewHydrator(db *gorm.DB, dir identity.Directory, backend backend.RecordTracker) *Hydrator {
+func NewHydrator(db *gorm.DB, dir identity.Directory, backend *backend.PostgresBackend) *Hydrator {
 	return &Hydrator{
 		db:      db,
 		dir:     dir,

hydration/post.go

···
 	ctx, span := tracer.Start(ctx, "hydratePost")
 	defer span.End()
 
-	// Query post from database
-	var dbPost models.Post
-	err := h.db.Raw(`SELECT * FROM posts
-			WHERE author = (SELECT id FROM repos WHERE did = ?)
-			AND rkey = ?
-	`, extractDIDFromURI(uri), extractRkeyFromURI(uri)).Scan(&dbPost).Error
+	autoFetch, _ := ctx.Value("auto-fetch").(bool)
 
+	authorDid := extractDIDFromURI(uri)
+	r, err := h.backend.GetOrCreateRepo(ctx, authorDid)
 	if err != nil {
+		return nil, err
+	}
+
+	// Query post from database
+	var dbPost models.Post
+	if err := h.db.Raw(`SELECT * FROM posts WHERE author = ? AND rkey = ? `, r.ID, extractRkeyFromURI(uri)).Scan(&dbPost).Error; err != nil {
 		return nil, fmt.Errorf("failed to query post: %w", err)
 	}
 
 	if dbPost.NotFound || len(dbPost.Raw) == 0 {
-		return nil, fmt.Errorf("post not found")
+		if autoFetch {
+			h.AddMissingRecord(uri, true)
+			if err := h.db.Raw(`SELECT * FROM posts WHERE author = ? AND rkey = ? `, r.ID, extractRkeyFromURI(uri)).Scan(&dbPost).Error; err != nil {
+				return nil, fmt.Errorf("failed to query post: %w", err)
+			}
+			if dbPost.NotFound || len(dbPost.Raw) == 0 {
+				return nil, fmt.Errorf("post not found")
+			}
+		} else {
+			return nil, fmt.Errorf("post not found")
+		}
 	}
 
 	// Unmarshal post record
···
 	// Get engagement counts
 	var likes, reposts, replies int
 	wg.Go(func() {
+		_, span := tracer.Start(ctx, "likeCounts")
+		defer span.End()
 		h.db.Raw("SELECT COUNT(*) FROM likes WHERE subject = ?", dbPost.ID).Scan(&likes)
 	})
 	wg.Go(func() {
+		_, span := tracer.Start(ctx, "repostCounts")
+		defer span.End()
 		h.db.Raw("SELECT COUNT(*) FROM reposts WHERE subject = ?", dbPost.ID).Scan(&reposts)
 	})
 	wg.Go(func() {
+		_, span := tracer.Start(ctx, "replyCounts")
+		defer span.End()
 		h.db.Raw("SELECT COUNT(*) FROM posts WHERE reply_to = ?", dbPost.ID).Scan(&replies)
 	})
 
···
 	var likeRkey string
 	if viewerDID != "" {
 		wg.Go(func() {
+			_, span := tracer.Start(ctx, "viewerLikeState")
+			defer span.End()
 			h.db.Raw(`
 			SELECT l.rkey FROM likes l
 			WHERE l.subject = ?
···
 	if embed == nil {
 		return nil
 	}
+	_, span := tracer.Start(ctx, "formatEmbed")
+	defer span.End()
 
 	result := &bsky.FeedDefs_PostView_Embed{}
 

main.go

···
 	"github.com/bluesky-social/indigo/util/cliutil"
 	xrpclib "github.com/bluesky-social/indigo/xrpc"
 	"github.com/gorilla/websocket"
-	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/ipfs/go-cid"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/urfave/cli/v2"
+	"github.com/whyrusleeping/konbini/backend"
 	"github.com/whyrusleeping/konbini/xrpc"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
···
 	"go.opentelemetry.io/otel/sdk/resource"
 	tracesdk "go.opentelemetry.io/otel/sdk/trace"
 	semconv "go.opentelemetry.io/otel/semconv/v1.20.0"
+	"gorm.io/gorm"
 	"gorm.io/gorm/logger"
 
 	. "github.com/whyrusleeping/konbini/models"
 )
-
-var handleOpHist = promauto.NewHistogramVec(prometheus.HistogramOpts{
-	Name:    "handle_op_duration",
-	Help:    "A histogram of op handling durations",
-	Buckets: prometheus.ExponentialBuckets(1, 2, 15),
-}, []string{"op", "collection"})
 
 var firehoseCursorGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
 	Name: "firehose_cursor",
···
 		db.AutoMigrate(Image{})
 		db.AutoMigrate(PostGate{})
 		db.AutoMigrate(StarterPack{})
-		db.AutoMigrate(SyncInfo{})
+		db.AutoMigrate(backend.SyncInfo{})
 		db.AutoMigrate(Notification{})
 		db.AutoMigrate(SequenceTracker{})
+		db.Exec("CREATE INDEX IF NOT EXISTS reposts_subject_idx ON reposts (subject)")
+		db.Exec("CREATE INDEX IF NOT EXISTS posts_reply_to_idx ON posts (reply_to)")
 
 		ctx := context.TODO()
 
-		rc, _ := lru.New2Q[string, *Repo](1_000_000)
-		pc, _ := lru.New2Q[string, cachedPostInfo](1_000_000)
-		revc, _ := lru.New2Q[uint, string](1_000_000)
-
 		cfg, err := pgxpool.ParseConfig(cctx.String("db-url"))
 		if err != nil {
 			return err
···
 			dir:    dir,
 
 			missingRecords: make(chan MissingRecord, 1024),
+			db:             db,
 		}
 		fmt.Println("MY DID: ", s.mydid)
 
-		pgb := &PostgresBackend{
-			relevantDids:  make(map[string]bool),
-			s:             s,
-			db:            db,
-			postInfoCache: pc,
-			repoCache:     rc,
-			revCache:      revc,
-			pgx:           pool,
+		pgb, err := backend.NewPostgresBackend(mydid, db, pool, cc, nil)
+		if err != nil {
+			return err
 		}
+
 		s.backend = pgb
 
-		myrepo, err := s.backend.getOrCreateRepo(ctx, mydid)
+		myrepo, err := s.backend.GetOrCreateRepo(ctx, mydid)
 		if err != nil {
 			return fmt.Errorf("failed to get repo record for our own did: %w", err)
 		}
 		s.myrepo = myrepo
 
-		if err := s.backend.loadRelevantDids(); err != nil {
+		if err := s.backend.LoadRelevantDids(); err != nil {
 			return fmt.Errorf("failed to load relevant dids set: %w", err)
 		}
 
···
 }
 
 type Server struct {
-	backend *PostgresBackend
+	backend *backend.PostgresBackend
 
 	dir identity.Directory
 
···
 
 	mpLk           sync.Mutex
 	missingRecords chan MissingRecord
+
+	db *gorm.DB
 }
 
 func (s *Server) getXrpcClient() (*xrpclib.Client, error) {
···
 				s.lastSeq = evt.Seq
 
 				if evt.Seq%1000 == 0 {
-					if err := storeLastSeq(s.backend.db, "firehose_seq", evt.Seq); err != nil {
+					if err := storeLastSeq(s.db, "firehose_seq", evt.Seq); err != nil {
 						fmt.Println("failed to store seqno: ", err)
 					}
 				}
···
 	return resp.DID.String(), nil
 }
 
-const (
-	NotifKindReply   = "reply"
-	NotifKindLike    = "like"
-	NotifKindMention = "mention"
-	NotifKindRepost  = "repost"
-)
-
-func (s *Server) AddNotification(ctx context.Context, forUser, author uint, recordUri string, recordCid cid.Cid, kind string) error {
-	return s.backend.db.Create(&Notification{
-		For:       forUser,
-		Author:    author,
-		Source:    recordUri,
-		SourceCid: recordCid.String(),
-		Kind:      kind,
-	}).Error
-}
-
 func (s *Server) rescanRepo(ctx context.Context, did string) error {
 	resp, err := s.dir.LookupDID(ctx, syntax.DID(did))
 	if err != nil {
 		return err
 	}
 
-	s.backend.addRelevantDid(did)
+	s.backend.AddRelevantDid(did)
 
 	c := &xrpclib.Client{
 		Host: resp.PDSEndpoint(),

missing.go

···
 }
 
 func (s *Server) fetchMissingProfile(ctx context.Context, did string) error {
-	s.backend.addRelevantDid(did)
+	s.backend.AddRelevantDid(did)
 
-	repo, err := s.backend.getOrCreateRepo(ctx, did)
+	repo, err := s.backend.GetOrCreateRepo(ctx, did)
 	if err != nil {
 		return err
 	}
···
 	collection := puri.Collection().String()
 	rkey := puri.RecordKey().String()
 
-	s.backend.addRelevantDid(did)
+	s.backend.AddRelevantDid(did)
 
-	repo, err := s.backend.getOrCreateRepo(ctx, did)
+	repo, err := s.backend.GetOrCreateRepo(ctx, did)
 	if err != nil {
 		return err
 	}
···
 	did := puri.Authority().String()
 	collection := puri.Collection().String()
 	rkey := puri.RecordKey().String()
-	s.backend.addRelevantDid(did)
+	s.backend.AddRelevantDid(did)
 
-	repo, err := s.backend.getOrCreateRepo(ctx, did)
+	repo, err := s.backend.GetOrCreateRepo(ctx, did)
 	if err != nil {
 		return err
 	}

pgbackend.go

···
-package main
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"strings"
-	"time"
-
-	"github.com/bluesky-social/indigo/api/atproto"
-	"github.com/bluesky-social/indigo/api/bsky"
-	"github.com/bluesky-social/indigo/atproto/syntax"
-	"github.com/bluesky-social/indigo/util"
-	"github.com/jackc/pgx/v5"
-	"github.com/jackc/pgx/v5/pgconn"
-	"github.com/whyrusleeping/market/models"
-	"gorm.io/gorm"
-	"gorm.io/gorm/clause"
-	"gorm.io/gorm/logger"
-
-	. "github.com/whyrusleeping/konbini/models"
-)
-
-func (b *PostgresBackend) DidToID(ctx context.Context, did string) (uint, error) {
-	r, err := b.getOrCreateRepo(ctx, did)
-	if err != nil {
-		return 0, err
-	}
-
-	return r.ID, nil
-}
-
-func (b *PostgresBackend) getOrCreateRepo(ctx context.Context, did string) (*Repo, error) {
-	r, ok := b.repoCache.Get(did)
-	if !ok {
-		b.reposLk.Lock()
-
-		r, ok = b.repoCache.Get(did)
-		if !ok {
-			r = &Repo{}
-			r.Did = did
-			b.repoCache.Add(did, r)
-		}
-
-		b.reposLk.Unlock()
-	}
-
-	r.Lk.Lock()
-	defer r.Lk.Unlock()
-	if r.Setup {
-		return r, nil
-	}
-
-	row := b.pgx.QueryRow(ctx, "SELECT id, created_at, did FROM repos WHERE did = $1", did)
-
-	err := row.Scan(&r.ID, &r.CreatedAt, &r.Did)
-	if err == nil {
-		// found it!
-		r.Setup = true
-		return r, nil
-	}
-
-	if err != pgx.ErrNoRows {
-		return nil, err
-	}
-
-	r.Did = did
-	if err := b.db.Create(r).Error; err != nil {
-		return nil, err
-	}
-
-	r.Setup = true
-
-	return r, nil
-}
-
-func (b *PostgresBackend) getOrCreateList(ctx context.Context, uri string) (*List, error) {
-	puri, err := util.ParseAtUri(uri)
-	if err != nil {
-		return nil, err
-	}
-
-	r, err := b.getOrCreateRepo(ctx, puri.Did)
-	if err != nil {
-		return nil, err
-	}
-
-	// TODO: needs upsert treatment when we actually find the list
-	var list List
-	if err := b.db.FirstOrCreate(&list, map[string]any{
-		"author": r.ID,
-		"rkey":   puri.Rkey,
-	}).Error; err != nil {
-		return nil, err
-	}
-	return &list, nil
-}
-
-type cachedPostInfo struct {
-	ID     uint
-	Author uint
-}
-
-func (b *PostgresBackend) postIDForUri(ctx context.Context, uri string) (uint, error) {
-	// getPostByUri implicitly fills the cache
-	p, err := b.postInfoForUri(ctx, uri)
-	if err != nil {
-		return 0, err
-	}
-
-	return p.ID, nil
-}
-
-func (b *PostgresBackend) postInfoForUri(ctx context.Context, uri string) (cachedPostInfo, error) {
-	v, ok := b.postInfoCache.Get(uri)
-	if ok {
-		return v, nil
-	}
-
-	// getPostByUri implicitly fills the cache
-	p, err := b.getOrCreatePostBare(ctx, uri)
-	if err != nil {
-		return cachedPostInfo{}, err
-	}
-
-	return cachedPostInfo{ID: p.ID, Author: p.Author}, nil
-}
-
-func (b *PostgresBackend) tryLoadPostInfo(ctx context.Context, uid uint, rkey string) (*Post, error) {
-	var p Post
-	q := "SELECT id, author FROM posts WHERE author = $1 AND rkey = $2"
-	if err := b.pgx.QueryRow(ctx, q, uid, rkey).Scan(&p.ID, &p.Author); err != nil {
-		if errors.Is(err, pgx.ErrNoRows) {
-			return nil, nil
-		}
-		return nil, err
-	}
-
-	return &p, nil
-}
-
-func (b *PostgresBackend) getOrCreatePostBare(ctx context.Context, uri string) (*Post, error) {
-	puri, err := util.ParseAtUri(uri)
-	if err != nil {
-		return nil, err
-	}
-
-	r, err := b.getOrCreateRepo(ctx, puri.Did)
-	if err != nil {
-		return nil, err
-	}
-
-	post, err := b.tryLoadPostInfo(ctx, r.ID, puri.Rkey)
-	if err != nil {
-		return nil, err
-	}
-
-	if post == nil {
-		post = &Post{
-			Rkey:     puri.Rkey,
-			Author:   r.ID,
-			NotFound: true,
-		}
-
-		err := b.pgx.QueryRow(ctx, "INSERT INTO posts (rkey, author, not_found) VALUES ($1, $2, $3) RETURNING id", puri.Rkey, r.ID, true).Scan(&post.ID)
-		if err != nil {
-			pgErr, ok := err.(*pgconn.PgError)
-			if !ok || pgErr.Code != "23505" {
-				return nil, err
-			}
-
-			out, err := b.tryLoadPostInfo(ctx, r.ID, puri.Rkey)
-			if err != nil {
-				return nil, fmt.Errorf("got duplicate post and still couldnt find it: %w", err)
-			}
-			if out == nil {
-				return nil, fmt.Errorf("postgres is lying to us: %d %s", r.ID, puri.Rkey)
-			}
-
-			post = out
-		}
-
-	}
-
-	b.postInfoCache.Add(uri, cachedPostInfo{
-		ID:     post.ID,
-		Author: post.Author,
-	})
-
-	return post, nil
-}
-
-func (b *PostgresBackend) getPostByUri(ctx context.Context, uri string, fields string) (*Post, error) {
-	puri, err := util.ParseAtUri(uri)
-	if err != nil {
-		return nil, err
-	}
-
-	r, err := b.getOrCreateRepo(ctx, puri.Did)
-	if err != nil {
-		return nil, err
-	}
-
-	q := "SELECT " + fields + " FROM posts WHERE author = ? AND rkey = ?"
-
-	var post Post
-	if err := b.db.Raw(q, r.ID, puri.Rkey).Scan(&post).Error; err != nil {
-		return nil, err
-	}
-
-	if post.ID == 0 {
-		post.Rkey = puri.Rkey
-		post.Author = r.ID
-		post.NotFound = true
-
-		if err := b.db.Session(&gorm.Session{
-			Logger: logger.Default.LogMode(logger.Silent),
-		}).Create(&post).Error; err != nil {
-			if !errors.Is(err, gorm.ErrDuplicatedKey) {
-				return nil, err
-			}
-			if err := b.db.Find(&post, "author = ? AND rkey = ?", r.ID, puri.Rkey).Error; err != nil {
-				return nil, fmt.Errorf("got duplicate post and still couldnt find it: %w", err)
-			}
-		}
-
-	}
-
-	b.postInfoCache.Add(uri, cachedPostInfo{
-		ID:     post.ID,
-		Author: post.Author,
-	})
-
-	return &post, nil
-}
-
-func (b *PostgresBackend) revForRepo(rr *Repo) (string, error) {
-	lrev, ok := b.revCache.Get(rr.ID)
-	if ok {
-		return lrev, nil
-	}
-
-	var rev string
-	if err := b.pgx.QueryRow(context.TODO(), "SELECT COALESCE(rev, '') FROM sync_infos WHERE repo = $1", rr.ID).Scan(&rev); err != nil {
-		if errors.Is(err, pgx.ErrNoRows) {
-			return "", nil
-		}
-		return "", err
-	}
-
-	if rev != "" {
-		b.revCache.Add(rr.ID, rev)
-	}
-	return rev, nil
-}
-
-func (b *PostgresBackend) ensureFollowsScraped(ctx context.Context, user string) error {
-	r, err := b.getOrCreateRepo(ctx, user)
-	if err != nil {
-		return err
-	}
-
-	var si SyncInfo
-	if err := b.db.Find(&si, "repo = ?", r.ID).Error; err != nil {
-		return err
-	}
-
-	// not found
-	if si.Repo == 0 {
-		if err := b.db.Create(&SyncInfo{
-			Repo: r.ID,
-		}).Error; err != nil {
-			return err
-		}
-	}
-
-	if si.FollowsSynced {
-		return nil
-	}
-
-	var follows []Follow
-	var cursor string
-	for {
-		resp, err := atproto.RepoListRecords(ctx, b.s.client, "app.bsky.graph.follow", cursor, 100, b.s.mydid, false)
-		if err != nil {
-			return err
-		}
-
-		for _, rec := range resp.Records {
-			if fol, ok := rec.Value.Val.(*bsky.GraphFollow); ok {
-				fr, err := b.getOrCreateRepo(ctx, fol.Subject)
-				if err != nil {
-					return err
-				}
-
-				puri, err := syntax.ParseATURI(rec.Uri)
-				if err != nil {
-					return err
-				}
-
-				follows = append(follows, Follow{
-					Created: time.Now(),
-					Indexed: time.Now(),
-					Rkey:    puri.RecordKey().String(),
-					Author:  r.ID,
-					Subject: fr.ID,
-				})
-			}
-		}
-
-		if resp.Cursor == nil || len(resp.Records) == 0 {
-			break
-		}
-		cursor = *resp.Cursor
-	}
-
-	if err := b.db.Clauses(clause.OnConflict{DoNothing: true}).CreateInBatches(follows, 200).Error; err != nil {
-		return err
-	}
-
-	if err := b.db.Model(SyncInfo{}).Where("repo = ?", r.ID).Update("follows_synced", true).Error; err != nil {
-		return err
-	}
-
-	fmt.Println("Got follows: ", len(follows))
-
-	return nil
-}
-
-func (b *PostgresBackend) loadRelevantDids() error {
-	ctx := context.TODO()
-
-	if err := b.ensureFollowsScraped(ctx, b.s.mydid); err != nil {
-		return fmt.Errorf("failed to scrape follows: %w", err)
-	}
-
-	r, err := b.getOrCreateRepo(ctx, b.s.mydid)
-	if err != nil {
-		return err
-	}
-
-	var dids []string
-	if err := b.db.Raw("select did from follows left join repos on follows.subject = repos.id where follows.author = ?", r.ID).Scan(&dids).Error; err != nil {
-		return err
-	}
-
-	b.relevantDids[b.s.mydid] = true
-	for _, d := range dids {
-		fmt.Println("adding did: ", d)
-		b.relevantDids[d] = true
-	}
-
-	return nil
-}
-
-type SyncInfo struct {
-	Repo          uint `gorm:"index"`
-	FollowsSynced bool
-	Rev           string
-}
-
-func (b *PostgresBackend) checkPostExists(ctx context.Context, repo *Repo, rkey string) (bool, error) {
-	var id uint
-	var notfound bool
-	if err := b.pgx.QueryRow(ctx, "SELECT id, not_found FROM posts WHERE author = $1 AND rkey = $2", repo.ID, rkey).Scan(&id, &notfound); err != nil {
-		if errors.Is(err, pgx.ErrNoRows) {
-			return false, nil
-		}
-		return false, err
-	}
-
-	if id != 0 && !notfound {
-		return true, nil
-	}
-
-	return false, nil
-}
-
-func (b *PostgresBackend) addRelevantDid(did string) {
-	b.rdLk.Lock()
-	defer b.rdLk.Unlock()
-	b.relevantDids[did] = true
-}
-
-func (b *PostgresBackend) didIsRelevant(did string) bool {
-	b.rdLk.Lock()
-	defer b.rdLk.Unlock()
-	return b.relevantDids[did]
-}
-
-func (b *PostgresBackend) anyRelevantIdents(idents ...string) bool {
-	for _, id := range idents {
-		if strings.HasPrefix(id, "did:") {
-			if b.didIsRelevant(id) {
-				return true
-			}
-		} else if strings.HasPrefix(id, "at://") {
-			puri, err := syntax.ParseATURI(id)
-			if err != nil {
-				continue
-			}
-
-			if b.didIsRelevant(puri.Authority().String()) {
-				return true
-			}
-		}
-	}
-
-	return false
-}
-
-func (b *PostgresBackend) getRepoByID(ctx context.Context, id uint) (*models.Repo, error) {
-	var r models.Repo
-	if err := b.db.Find(&r, "id = ?", id).Error; err != nil {
-		return nil, err
-	}
-
-	return &r, nil
-}
-
-func (b *PostgresBackend) TrackMissingRecord(identifier string, wait bool) {
-	b.s.addMissingRecord(context.TODO(), MissingRecord{
-		Type:       inferRecordType(identifier),
-		Identifier: identifier,
-		Wait:       wait,
-	})
-}
-
-// inferRecordType determines the record type based on the identifier format
-func inferRecordType(identifier string) MissingRecordType {
-	if strings.HasPrefix(identifier, "did:") {
-		return MissingRecordTypeProfile
-	}
-
-	if strings.HasPrefix(identifier, "at://") {
-		if strings.Contains(identifier, "/app.bsky.feed.post/") {
-			return MissingRecordTypePost
-		}
-		if strings.Contains(identifier, "/app.bsky.feed.generator/") {
-			return MissingRecordTypeFeedGenerator
-		}
-	}
-
-	// Default to post if we can't determine
-	return MissingRecordTypePost
-}

xrpc/feed/getAuthorFeed.go

···
 	"time"
 
 	"github.com/bluesky-social/indigo/api/bsky"
+	"github.com/bluesky-social/indigo/atproto/syntax"
 	"github.com/labstack/echo/v4"
 	"github.com/whyrusleeping/konbini/hydration"
 	"github.com/whyrusleeping/konbini/views"
···
 		go func(i int, row postRow) {
 			defer wg.Done()
 
-			postInfo, err := hydrator.HydratePost(ctx, row.URI, viewer)
+			puri, err := syntax.ParseATURI(row.URI)
 			if err != nil {
-				if strings.Contains(err.Error(), "post not found") {
-					hydrator.AddMissingRecord(row.URI, true)
-					postInfo, err = hydrator.HydratePost(ctx, row.URI, viewer)
-					if err != nil {
-						slog.Error("failed to hydrate post after fetch missing", "uri", row.URI, "error", err)
+				slog.Error("row had invalid uri", "uri", row.URI, "error", err)
+				return
+			}
+
+			var subwg sync.WaitGroup
+
+			var postInfo *hydration.PostInfo
+			subwg.Go(func() {
+				pi, err := hydrator.HydratePost(ctx, row.URI, viewer)
+				if err != nil {
+					if strings.Contains(err.Error(), "post not found") {
+						hydrator.AddMissingRecord(row.URI, true)
+						pi, err = hydrator.HydratePost(ctx, row.URI, viewer)
+						if err != nil {
+							slog.Error("failed to hydrate post after fetch missing", "uri", row.URI, "error", err)
+							return
+						}
+					} else {
+						slog.Warn("failed to hydrate post", "uri", row.URI, "error", err)
 						return
 					}
-				} else {
-					slog.Warn("failed to hydrate post", "uri", row.URI, "error", err)
+				}
+				postInfo = pi
+			})
+
+			var authorInfo *hydration.ActorInfo
+			subwg.Go(func() {
+				ai, err := hydrator.HydrateActor(ctx, puri.Authority().String())
+				if err != nil {
+					hydrator.AddMissingRecord(postInfo.Author, false)
+					slog.Warn("failed to hydrate author", "did", postInfo.Author, "error", err)
 					return
 				}
-			}
+				authorInfo = ai
+			})
 
-			authorInfo, err := hydrator.HydrateActor(ctx, postInfo.Author)
-			if err != nil {
-				hydrator.AddMissingRecord(postInfo.Author, false)
-				slog.Warn("failed to hydrate author", "did", postInfo.Author, "error", err)
+			subwg.Wait()
+
+			if postInfo == nil || authorInfo == nil {
 				return
 			}
 

xrpc/server.go

···
 package xrpc
 
 import (
+	"context"
 	"log/slog"
 	"net/http"
 
 	"github.com/bluesky-social/indigo/atproto/identity"
 	"github.com/labstack/echo/v4"
 	"github.com/labstack/echo/v4/middleware"
+	"github.com/whyrusleeping/konbini/backend"
 	"github.com/whyrusleeping/konbini/hydration"
+	"github.com/whyrusleeping/konbini/models"
 	"github.com/whyrusleeping/konbini/xrpc/actor"
 	"github.com/whyrusleeping/konbini/xrpc/feed"
 	"github.com/whyrusleeping/konbini/xrpc/graph"
···
 	// Add methods as needed for data access
 
 	TrackMissingRecord(identifier string, wait bool)
+	GetOrCreateRepo(ctx context.Context, did string) (*models.Repo, error)
 }
 
 // NewServer creates a new XRPC server
-func NewServer(db *gorm.DB, dir identity.Directory, backend Backend) *Server {
+func NewServer(db *gorm.DB, dir identity.Directory, backend *backend.PostgresBackend) *Server {
 	e := echo.New()
 	e.HidePort = true
 	e.HideBanner = true

xrpc/unspecced/getPostThreadV2.go

···
 // HandleGetPostThreadV2 implements app.bsky.unspecced.getPostThreadV2
 func HandleGetPostThreadV2(c echo.Context, db *gorm.DB, hydrator *hydration.Hydrator) error {
 	ctx := c.Request().Context()
+	ctx = context.WithValue(ctx, "auto-fetch", true)
 
 	// Parse parameters
 	anchorRaw := c.QueryParam("anchor")