package appview

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"maps"
	"net/http"
	"net/url"
	"slices"
	"strings"
	"sync"

	"time"

	"github.com/avast/retry-go/v4"
	"github.com/bluesky-social/indigo/atproto/syntax"
	jmodels "github.com/bluesky-social/jetstream/pkg/models"
	"github.com/go-git/go-git/v5/plumbing"
	"github.com/ipfs/go-cid"
	"golang.org/x/sync/errgroup"
	"tangled.org/core/api/tangled"
	"tangled.org/core/appview/cache"
	"tangled.org/core/appview/config"
	"tangled.org/core/appview/db"
	"tangled.org/core/appview/knotacl"
	"tangled.org/core/appview/mentions"
	"tangled.org/core/appview/models"
	"tangled.org/core/appview/notify"
	"tangled.org/core/appview/serververify"
	"tangled.org/core/consts"
	"tangled.org/core/idresolver"
	"tangled.org/core/orm"
	"tangled.org/core/rbac"
	"tangled.org/core/repoverify"
)

type RepoPermissionChecker interface {
	HasRepoPermissionErr(ctx context.Context, repo *models.Repo, userDid, perm string) (bool, error)
}

type Ingester struct {
	Ctx              context.Context
	Db               *db.DB
	Enforcer         *rbac.Enforcer
	Acl              RepoPermissionChecker
	IdResolver       *idresolver.Resolver
	Cache            *cache.Cache
	Config           *config.Config
	Logger           *slog.Logger
	MentionsResolver *mentions.Resolver
	Notifier         notify.Notifier
	Verifier         repoverify.Verifier
}

type processFunc func(ctx context.Context, e *jmodels.Event) error

func (i *Ingester) Ingest() processFunc {
	return func(ctx context.Context, e *jmodels.Event) error {
		var err error

		l := i.Logger.With("kind", e.Kind)
		switch e.Kind {
		case jmodels.EventKindAccount:
			// TODO: sync account state to db
			if e.Account.Active {
				break
			}
			// TODO: revoke sessions by DID
			if *e.Account.Status == "deactivated" {
				err = i.IdResolver.InvalidateIdent(ctx, e.Account.Did)
			}
		case jmodels.EventKindIdentity:
			err = i.IdResolver.InvalidateIdent(ctx, e.Identity.Did)
		case jmodels.EventKindCommit:
			l = l.With(
				"nsid", e.Commit.Collection,
				"did", e.Did,
				"rkey", e.Commit.RKey,
				"op", e.Commit.Operation,
			)
			switch e.Commit.Collection {
			case tangled.GraphFollowNSID:
				err = i.ingestFollow(e, l)
			case tangled.GraphVouchNSID:
				err = i.ingestVouch(ctx, e, l)
			case tangled.FeedStarNSID:
				err = i.ingestStar(ctx, e, l)
			case tangled.FeedReactionNSID:
				err = i.ingestReaction(e, l)
			case tangled.PublicKeyNSID:
				err = i.ingestPublicKey(e, l)
			case tangled.RepoArtifactNSID:
				err = i.ingestArtifact(ctx, e, l)
			case tangled.ActorProfileNSID:
				err = i.ingestProfile(ctx, e, l)
			case tangled.SpindleMemberNSID:
				err = i.ingestSpindleMember(ctx, e, l)
			case tangled.SpindleNSID:
				err = i.ingestSpindle(ctx, e, l)
			case tangled.KnotMemberNSID:
				err = i.ingestKnotMember(ctx, e, l)
			case tangled.KnotNSID:
				err = i.ingestKnot(ctx, e, l)
			case tangled.StringNSID:
				err = i.ingestString(e, l)
			case tangled.RepoIssueNSID:
				err = i.ingestIssue(ctx, e, l)
			case tangled.RepoIssueStateNSID:
				err = i.ingestState(ctx, e, l, issueStateSpec)
			case tangled.RepoPullNSID:
				err = i.ingestPull(ctx, e, l)
			case tangled.RepoPullStatusNSID:
				err = i.ingestState(ctx, e, l, pullStatusSpec)
			case tangled.FeedCommentNSID:
				err = i.ingestComment(e, l)
			case tangled.RepoIssueCommentNSID:
				err = i.ingestIssueComment(e, l)
			case tangled.RepoPullCommentNSID:
				err = i.ingestPullComment(e, l)
			case tangled.LabelDefinitionNSID:
				err = i.ingestLabelDefinition(e, l)
			case tangled.LabelOpNSID:
				err = i.ingestLabelOp(ctx, e, l)
			case tangled.RepoNSID:
				err = i.ingestRepo(ctx, e, l)
			}
		}

		if err != nil {
			l.Warn("failed to ingest record, skipping", "err", err)
		}

		lastTimeUs := e.TimeUS + 1
		if saveErr := i.Db.SaveLastTimeUs(lastTimeUs); saveErr != nil {
			l.Error("failed to save cursor", "err", saveErr)
		}

		return nil
	}
}

func (i *Ingester) resolveRepoRef(ref string) (*models.Repo, error) {
	if strings.HasPrefix(ref, "did:") {
		return db.GetRepoByDid(i.Db, ref)
	}
	return db.GetRepoByAtUri(i.Db, ref)
}

func (i *Ingester) resolveOldFormatStar(raw json.RawMessage, star *models.Star, l *slog.Logger) (bool, error) {
	var legacy struct {
		Subject    *string `json:"subject"`
		SubjectDid *string `json:"subjectDid"`
	}
	if err := json.Unmarshal(raw, &legacy); err != nil {
		return false, err
	}

	switch {
	case legacy.SubjectDid != nil:
		repo, err := i.resolveRepoRef(*legacy.SubjectDid)
		if err != nil {
			l.Warn("skipping old-format star for unknown repo", "subjectDid", *legacy.SubjectDid)
			return false, nil
		}
		star.SubjectType = models.StarSubjectRepo
		star.Subject = repo.RepoDid
		return true, nil

	case legacy.Subject != nil:
		uri, err := syntax.ParseATURI(*legacy.Subject)
		if err != nil {
			return false, fmt.Errorf("invalid old-format star subject: %w", err)
		}
		switch uri.Collection().String() {
		case tangled.RepoNSID:
			repo, err := db.GetRepoByAtUri(i.Db, uri.String())
			if err != nil {
				l.Warn("skipping old-format star for unknown repo", "subject", *legacy.Subject)
				return false, nil
			}
			star.SubjectType = models.StarSubjectRepo
			star.Subject = repo.RepoDid
			return true, nil
		default:
			star.SubjectType = models.StarSubjectString
			star.Subject = *legacy.Subject
			return true, nil
		}

	default:
		return false, fmt.Errorf("old-format star has neither subject nor subjectDid")
	}
}

func (i *Ingester) ingestStar(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {
	var err error
	did := e.Did

	l = l.With("handler", "ingestStar")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.FeedStar{}
		unmarshalErr := json.Unmarshal(raw, &record)

		createdAt, parseErr := time.Parse(time.RFC3339, record.CreatedAt)
		if parseErr != nil {
			createdAt = time.Now()
		}

		star := models.Star{
			Did:     did,
			Rkey:    e.Commit.RKey,
			Created: createdAt,
		}

		switch {
		case unmarshalErr != nil:
			resolved, resolveErr := i.resolveOldFormatStar(raw, &star, l)
			if resolveErr != nil {
				l.Error("invalid record", "newFmtErr", unmarshalErr, "oldFmtErr", resolveErr)
				return unmarshalErr
			}
			if !resolved {
				return nil
			}

		case record.Subject == nil:
			return fmt.Errorf("star record has nil subject")

		case record.Subject.FeedStar_Repo != nil:
			repo, repoErr := i.resolveRepoRef(record.Subject.FeedStar_Repo.Did)
			if repoErr != nil {
				l.Warn("skipping star for unknown repo", "did", record.Subject.FeedStar_Repo.Did)
				return nil
			}
			star.SubjectType = models.StarSubjectRepo
			star.Subject = repo.RepoDid

		case record.Subject.FeedStar_String != nil:
			star.SubjectType = models.StarSubjectString
			star.Subject = record.Subject.FeedStar_String.Uri

		default:
			return fmt.Errorf("star record has empty subject union")
		}

		err = db.UpsertStar(i.Db, star)
	case jmodels.CommitOperationDelete:
		err = db.DeleteStarByRkey(i.Db, did, e.Commit.RKey)
	}

	if err != nil {
		return fmt.Errorf("failed to %s star record: %w", e.Commit.Operation, err)
	}
	l.Info("processed star", "operation", e.Commit.Operation, "rkey", e.Commit.RKey)

	l.Info("ingested record")
	return nil
}

func (i *Ingester) ingestFollow(e *jmodels.Event, l *slog.Logger) error {
	var err error
	did := e.Did

	l = l.With("handler", "ingestFollow")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.GraphFollow{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		err = db.UpsertFollow(i.Db, models.Follow{
			UserDid:    did,
			SubjectDid: record.Subject,
			Rkey:       e.Commit.RKey,
		})
	case jmodels.CommitOperationDelete:
		err = db.DeleteFollowByRkey(i.Db, did, e.Commit.RKey)
	}

	if err != nil {
		return fmt.Errorf("failed to %s follow record: %w", e.Commit.Operation, err)
	}
	l.Info("processed follow", "operation", e.Commit.Operation, "rkey", e.Commit.RKey)

	l.Info("ingested record")
	return nil
}

func (i *Ingester) ingestVouch(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {
	var err error
	did := e.Did

	l = l.With("handler", "ingestVouch")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.GraphVouch{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		// rkey is the subject_did being vouched for/denounced
		subjectDID := e.Commit.RKey

		_, err = syntax.ParseDID(subjectDID)
		if err != nil {
			l.Error("invalid subject_did in rkey", "err", err, "rkey", subjectDID)
			return fmt.Errorf("invalid subject_did: %w", err)
		}

		if did == subjectDID {
			l.Warn("attempted self-vouch", "did", did)
			return fmt.Errorf("cannot vouch for self")
		}

		subjectId, err := i.IdResolver.ResolveIdent(ctx, subjectDID)
		if err != nil {
			return err
		}

		if subjectId.Handle.IsInvalidHandle() {
			return err
		}

		kind, err := models.ParseVouchKind(record.Kind)
		if err != nil {
			l.Error("invalid kind", "kind", kind)
			return fmt.Errorf("invalid kind: %s", kind)
		}

		recordCid, err := cid.Parse(e.Commit.CID)
		if err != nil {
			l.Error("invalid cid", "err", err, "cid", e.Commit.CID)
			return fmt.Errorf("invalid cid: %w", err)
		}

		var evidences []syntax.ATURI
		for _, raw := range record.Evidences {
			uri, parseErr := syntax.ParseATURI(raw)
			if parseErr != nil {
				l.Warn("invalid evidence AT-URI, skipping", "uri", raw, "err", parseErr)
				continue
			}
			evidences = append(evidences, uri)
		}

		tx, txErr := i.Db.Begin()
		if txErr != nil {
			return fmt.Errorf("failed to start transaction: %w", txErr)
		}

		addErr := db.AddVouch(tx, &models.Vouch{
			Did:        syntax.DID(did),
			SubjectDid: subjectId.DID,
			Cid:        recordCid,
			Kind:       kind,
			Reason:     record.Reason,
			Evidences:  evidences,
		})
		if addErr != nil {
			tx.Rollback()
			err = addErr
		} else {
			err = tx.Commit()
		}

	case jmodels.CommitOperationDelete:
		err = db.DeleteVouchByRkey(i.Db, did, e.Commit.RKey)
	}

	if err != nil {
		return fmt.Errorf("failed to %s vouch record: %w", e.Commit.Operation, err)
	}

	l.Info("ingested record")
	return nil
}

func (i *Ingester) ingestPublicKey(e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	var err error

	l = l.With("handler", "ingestPublicKey")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		l.Debug("processing add of pubkey")
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.PublicKey{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}
		pubKey, err := models.PublicKeyFromRecord(syntax.DID(did), syntax.RecordKey(e.Commit.RKey), record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}
		if err := pubKey.Validate(); err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		err = db.UpsertPublicKey(i.Db, pubKey)
	case jmodels.CommitOperationDelete:
		l.Debug("processing delete of pubkey")
		err = db.DeletePublicKeyByRkey(i.Db, did, e.Commit.RKey)
	}

	if err != nil {
		return fmt.Errorf("failed to %s pubkey record: %w", e.Commit.Operation, err)
	}
	l.Info("processed pubkey", "operation", e.Commit.Operation, "rkey", e.Commit.RKey)

	l.Info("ingested record")
	return nil
}

func (i *Ingester) ingestArtifact(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	var err error

	l = l.With("handler", "ingestArtifact")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.RepoArtifact{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		var repo *models.Repo
		if record.RepoDid != nil && *record.RepoDid != "" {
			repo, err = db.GetRepoByDid(i.Db, *record.RepoDid)
			if err != nil && !errors.Is(err, sql.ErrNoRows) {
				return fmt.Errorf("failed to look up repo by DID %s: %w", *record.RepoDid, err)
			}
		}
		if repo == nil && record.Repo != nil {
			repoAt, parseErr := syntax.ParseATURI(*record.Repo)
			if parseErr != nil {
				return parseErr
			}
			repo, err = db.GetRepoByAtUri(i.Db, repoAt.String())
			if err != nil {
				return err
			}
		}
		if repo == nil {
			return fmt.Errorf("artifact record has neither valid repoDid nor repo field")
		}

		allowed, permErr := i.Acl.HasRepoPermissionErr(ctx, repo, did, "repo:push")
		if permErr != nil {
			l.Warn("ingesting artifact without permission check", "did", did, "repo", repo.RepoIdentifier(), "err", permErr)
		} else if !allowed {
			l.Info("skipping unauthorized artifact", "did", did, "repo", repo.RepoIdentifier())
			return nil
		}

		repoDid := repo.RepoDid
		if repoDid == "" && record.RepoDid != nil {
			repoDid = *record.RepoDid
		}
		if repoDid != "" && (record.RepoDid == nil || *record.RepoDid == "") && record.Repo != nil {
			if enqErr := db.EnqueuePdsRecordMigration(ctx, i.Db, "add-repo-did", syntax.DID(did), syntax.NSID(tangled.RepoArtifactNSID), syntax.RecordKey(e.Commit.RKey)); enqErr != nil {
				l.Warn("failed to enqueue PDS rewrite for artifact", "err", enqErr, "did", did, "repoDid", repoDid)
			}
		}

		createdAt, parseErr := time.Parse(time.RFC3339, record.CreatedAt)
		if parseErr != nil {
			createdAt = time.Now()
		}

		artifact := models.Artifact{
			Did:       did,
			Rkey:      e.Commit.RKey,
			RepoDid:   syntax.DID(repo.RepoDid),
			Tag:       plumbing.Hash(record.Tag),
			CreatedAt: createdAt,
			BlobCid:   cid.Cid(record.Artifact.Ref),
			Name:      record.Name,
			Size:      uint64(record.Artifact.Size),
			MimeType:  record.Artifact.MimeType,
		}

		err = db.AddArtifact(i.Db, artifact)
	case jmodels.CommitOperationDelete:
		err = db.DeleteArtifact(i.Db, orm.FilterEq("did", did), orm.FilterEq("rkey", e.Commit.RKey))
	}

	if err != nil {
		return fmt.Errorf("failed to %s artifact record: %w", e.Commit.Operation, err)
	}

	l.Info("ingested record")
	return nil
}

func (i *Ingester) ingestProfile(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	var err error

	l = l.With("handler", "ingestProfile")

	if e.Commit.RKey != "self" {
		return fmt.Errorf("ingestProfile only ingests `self` record")
	}

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.ActorProfile{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		avatar := ""
		if record.Avatar != nil {
			avatar = record.Avatar.Ref.String()
		}

		description := ""
		if record.Description != nil {
			description = *record.Description
		}

		includeBluesky := record.Bluesky

		pronouns := ""
		if record.Pronouns != nil {
			pronouns = *record.Pronouns
		}

		location := ""
		if record.Location != nil {
			location = *record.Location
		}

		var links [5]string
		for i, l := range record.Links {
			if i < 5 {
				links[i] = l
			}
		}

		var stats [2]models.VanityStat
		for i, s := range record.Stats {
			if i < 2 {
				stats[i].Kind = models.ParseVanityStatKind(s)
			}
		}

		var pinned [6]string
		for i, r := range record.PinnedRepositories {
			if i < 6 {
				pinned[i] = r
			}
		}

		var preferredHandle syntax.Handle
		if record.PreferredHandle != nil {
			if h, err := syntax.ParseHandle(*record.PreferredHandle); err == nil {
				ident, identErr := i.IdResolver.ResolveIdent(ctx, did)
				if identErr == nil && slices.Contains(ident.AlsoKnownAs, "at://"+string(h)) {
					preferredHandle = h
				}
			}
		}

		profile := models.Profile{
			Did:             did,
			Avatar:          avatar,
			Description:     description,
			IncludeBluesky:  includeBluesky,
			Location:        location,
			Links:           links,
			Stats:           stats,
			PinnedRepos:     pinned,
			Pronouns:        pronouns,
			PreferredHandle: preferredHandle,
		}

		err = db.ValidateProfile(i.Db, &profile)
		if err != nil {
			return fmt.Errorf("invalid profile record")
		}

		err = db.UpsertProfile(i.Db, &profile)
		if err != nil {
			return fmt.Errorf("upserting profile: %w", err)
		}

		if i.Cache != nil {
			pipe := i.Cache.Pipeline()
			didKey := fmt.Sprintf(cache.PreferredHandleByDid, did)
			if preferredHandle != "" {
				pipe.Set(ctx, didKey, string(preferredHandle), cache.PreferredHandleTTL)
				pipe.Set(ctx, fmt.Sprintf(cache.PreferredHandleByHandle, string(preferredHandle)), did, cache.PreferredHandleTTL)
			} else {
				pipe.Del(ctx, didKey)
			}
			if _, execErr := pipe.Exec(ctx); execErr != nil {
				l.Warn("failed to update preferred handle cache", "err", execErr)
			}
		}
	case jmodels.CommitOperationDelete:
		tx, beginErr := i.Db.Begin()
		if beginErr != nil {
			return fmt.Errorf("failed to start transaction: %w", beginErr)
		}

		priorHandle, phErr := db.GetPreferredHandle(tx, did)
		if phErr != nil && !errors.Is(phErr, sql.ErrNoRows) {
			l.Warn("failed to read prior preferred handle", "err", phErr)
		}

		err = db.DeleteProfile(tx, did)
		if err == nil && i.Cache != nil {
			pipe := i.Cache.Pipeline()
			pipe.Del(ctx, fmt.Sprintf(cache.PreferredHandleByDid, did))
			if priorHandle != "" {
				pipe.Del(ctx, fmt.Sprintf(cache.PreferredHandleByHandle, string(priorHandle)))
			}
			if _, execErr := pipe.Exec(ctx); execErr != nil {
				l.Warn("failed to evict preferred handle cache", "err", execErr)
			}
		}
	}

	if err != nil {
		return fmt.Errorf("failed to %s profile record: %w", e.Commit.Operation, err)
	}

	l.Info("ingested record")
	return nil
}

func (i *Ingester) ingestSpindleMember(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	var err error

	l = l.With("handler", "ingestSpindleMember")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.SpindleMember{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		// only spindle owner can invite to spindles
		ok, err := i.Enforcer.IsSpindleInviteAllowed(did, record.Instance)
		if err != nil {
			return fmt.Errorf("failed to check invite permission: %w", err)
		}
		if !ok {
			if verifyErr := i.verifySpindle(ctx, record.Instance, did); verifyErr != nil {
				return fmt.Errorf("invite denied and verify failed: %w", verifyErr)
			}
			ok, err = i.Enforcer.IsSpindleInviteAllowed(did, record.Instance)
			if err != nil {
				return fmt.Errorf("failed to re-check invite permission: %w", err)
			}
			if !ok {
				return fmt.Errorf("invite denied for did %s on spindle %s", did, record.Instance)
			}
		}

		memberId, err := i.IdResolver.ResolveIdent(ctx, record.Subject)
		if err != nil {
			return err
		}

		if memberId.Handle.IsInvalidHandle() {
			return fmt.Errorf("invalid handle for member %s", record.Subject)
		}

		existing, err := db.GetSpindleMembers(i.Db,
			orm.FilterEq("did", did),
			orm.FilterEq("rkey", e.Commit.RKey),
		)
		if err != nil {
			return fmt.Errorf("failed to look up existing member: %w", err)
		}
		if len(existing) > 1 {
			return fmt.Errorf("multiple spindle members with rkey %s", e.Commit.RKey)
		}

		tx, err := i.Db.Begin()
		if err != nil {
			return fmt.Errorf("failed to start txn: %w", err)
		}
		committed := false
		defer func() {
			if committed {
				return
			}
			tx.Rollback()
			i.Enforcer.E.LoadPolicy()
		}()

		if len(existing) == 1 {
			prev := existing[0]
			if prev.Instance != record.Instance || prev.Subject != memberId.DID {
				if err = db.RemoveSpindleMember(tx,
					orm.FilterEq("did", did),
					orm.FilterEq("rkey", e.Commit.RKey),
				); err != nil {
					return fmt.Errorf("failed to remove stale row: %w", err)
				}
				if err = i.Enforcer.RemoveSpindleMember(prev.Instance, prev.Subject.String()); err != nil {
					return fmt.Errorf("failed to remove stale ACL: %w", err)
				}
			}
		}

		if err = db.AddSpindleMember(tx, models.SpindleMember{
			Did:      syntax.DID(did),
			Rkey:     e.Commit.RKey,
			Instance: record.Instance,
			Subject:  memberId.DID,
		}); err != nil {
			return fmt.Errorf("failed to add to db: %w", err)
		}

		if err = i.Enforcer.AddSpindleMember(record.Instance, memberId.DID.String()); err != nil {
			return fmt.Errorf("failed to update ACLs: %w", err)
		}

		if err = tx.Commit(); err != nil {
			return fmt.Errorf("failed to commit txn: %w", err)
		}

		if err = i.Enforcer.E.SavePolicy(); err != nil {
			return fmt.Errorf("failed to save ACLs: %w", err)
		}
		committed = true

		l.Info("upserted spindle member")
	case jmodels.CommitOperationDelete:
		rkey := e.Commit.RKey

		// get record from db first
		members, err := db.GetSpindleMembers(
			i.Db,
			orm.FilterEq("did", did),
			orm.FilterEq("rkey", rkey),
		)
		if err != nil || len(members) != 1 {
			return fmt.Errorf("failed to get member: %w, len(members) = %d", err, len(members))
		}
		member := members[0]

		tx, err := i.Db.Begin()
		if err != nil {
			return fmt.Errorf("failed to start txn: %w", err)
		}
		committed := false
		defer func() {
			if committed {
				return
			}
			tx.Rollback()
			i.Enforcer.E.LoadPolicy()
		}()

		// remove record by rkey && update enforcer
		if err = db.RemoveSpindleMember(
			tx,
			orm.FilterEq("did", did),
			orm.FilterEq("rkey", rkey),
		); err != nil {
			return fmt.Errorf("failed to remove from db: %w", err)
		}

		// update enforcer
		err = i.Enforcer.RemoveSpindleMember(member.Instance, member.Subject.String())
		if err != nil {
			return fmt.Errorf("failed to update ACLs: %w", err)
		}

		if err = tx.Commit(); err != nil {
			return fmt.Errorf("failed to commit txn: %w", err)
		}

		if err = i.Enforcer.E.SavePolicy(); err != nil {
			return fmt.Errorf("failed to save ACLs: %w", err)
		}
		committed = true

		l.Info("removed spindle member")
	}

	return nil
}

func (i *Ingester) ingestSpindle(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	var err error

	l = l.With("handler", "ingestSpindle")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.Spindle{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		instance := e.Commit.RKey

		err := db.AddSpindle(i.Db, models.Spindle{
			Owner:    syntax.DID(did),
			Instance: instance,
		})
		if err != nil {
			l.Error("failed to add spindle to db", "err", err, "instance", instance)
			return err
		}

		if err := i.verifySpindle(ctx, instance, did); err != nil {
			l.Warn("failed to verify spindle", "instance", instance, "did", did, "err", err)
		}

		l.Info("ingested record", "instance", instance)
		return nil

	case jmodels.CommitOperationDelete:
		instance := e.Commit.RKey

		// get record from db first
		spindles, err := db.GetSpindles(
			ctx,
			i.Db,
			orm.FilterEq("owner", did),
			orm.FilterEq("instance", instance),
		)
		if err != nil || len(spindles) != 1 {
			return fmt.Errorf("failed to get spindles: %w, len(spindles) = %d", err, len(spindles))
		}
		spindle := spindles[0]

		tx, err := i.Db.Begin()
		if err != nil {
			return fmt.Errorf("failed to start txn: %w", err)
		}
		defer func() {
			tx.Rollback()
			i.Enforcer.E.LoadPolicy()
		}()

		// remove spindle members first
		err = db.RemoveSpindleMember(
			tx,
			orm.FilterEq("owner", did),
			orm.FilterEq("instance", instance),
		)
		if err != nil {
			return fmt.Errorf("failed to remove spindle members: %w", err)
		}

		err = db.DeleteSpindle(
			tx,
			orm.FilterEq("owner", did),
			orm.FilterEq("instance", instance),
		)
		if err != nil {
			return fmt.Errorf("failed to delete spindle: %w", err)
		}

		if spindle.Verified != nil {
			err = i.Enforcer.RemoveSpindle(instance)
			if err != nil {
				return fmt.Errorf("failed to remove spindle from enforcer: %w", err)
			}
		}

		err = tx.Commit()
		if err != nil {
			return fmt.Errorf("failed to commit txn: %w", err)
		}

		err = i.Enforcer.E.SavePolicy()
		if err != nil {
			return fmt.Errorf("failed to save ACLs: %w", err)
		}

		l.Info("ingested record", "instance", instance)
	}

	return nil
}

func (i *Ingester) ingestString(e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	rkey := e.Commit.RKey

	var err error

	l = l.With("handler", "ingestString")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.String{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		string := models.StringFromRecord(did, rkey, record)

		if err = string.Validate(); err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		if err = db.AddString(i.Db, string); err != nil {
			l.Error("failed to add string", "err", err)
			return err
		}

		l.Info("ingested record")
		return nil

	case jmodels.CommitOperationDelete:
		if err := db.DeleteString(
			i.Db,
			orm.FilterEq("did", did),
			orm.FilterEq("rkey", rkey),
		); err != nil {
			l.Error("failed to delete", "err", err)
			return fmt.Errorf("failed to delete string record: %w", err)
		}

		l.Info("ingested record")
		return nil
	}

	return nil
}

func (i *Ingester) ingestKnotMember(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	var err error

	l = l.With("handler", "ingestKnotMember")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.KnotMember{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		// only knot owner can invite to knots
		ok, err := i.Enforcer.IsKnotInviteAllowed(did, record.Domain)
		if err != nil {
			return fmt.Errorf("failed to check invite permission: %w", err)
		}
		if !ok {
			if verifyErr := i.verifyKnot(ctx, record.Domain, did); verifyErr != nil {
				return fmt.Errorf("invite denied and verify failed: %w", verifyErr)
			}
			ok, err = i.Enforcer.IsKnotInviteAllowed(did, record.Domain)
			if err != nil {
				return fmt.Errorf("failed to re-check invite permission: %w", err)
			}
			if !ok {
				return fmt.Errorf("invite denied for did %s on knot %s", did, record.Domain)
			}
		}

		memberId, err := i.IdResolver.ResolveIdent(ctx, record.Subject)
		if err != nil {
			return err
		}

		if memberId.Handle.IsInvalidHandle() {
			return fmt.Errorf("invalid handle for member %s", record.Subject)
		}

		existing, err := db.GetKnotMembers(i.Db,
			orm.FilterEq("did", did),
			orm.FilterEq("rkey", e.Commit.RKey),
		)
		if err != nil {
			return fmt.Errorf("failed to look up existing member: %w", err)
		}
		if len(existing) > 1 {
			return fmt.Errorf("multiple knot members with rkey %s", e.Commit.RKey)
		}

		tx, err := i.Db.Begin()
		if err != nil {
			return fmt.Errorf("failed to start txn: %w", err)
		}
		committed := false
		defer func() {
			if committed {
				return
			}
			tx.Rollback()
			i.Enforcer.E.LoadPolicy()
		}()

		if len(existing) == 1 {
			prev := existing[0]
			if prev.Domain != record.Domain || prev.Subject != memberId.DID {
				if err = db.RemoveKnotMember(tx,
					orm.FilterEq("did", did),
					orm.FilterEq("rkey", e.Commit.RKey),
				); err != nil {
					return fmt.Errorf("failed to remove stale row: %w", err)
				}
				if err = i.Enforcer.RemoveKnotMember(prev.Domain, prev.Subject.String()); err != nil {
					return fmt.Errorf("failed to remove stale ACL: %w", err)
				}
			}
		}

		if err = db.AddKnotMember(tx, models.KnotMember{
			Did:     syntax.DID(did),
			Rkey:    e.Commit.RKey,
			Domain:  record.Domain,
			Subject: memberId.DID,
		}); err != nil {
			return fmt.Errorf("failed to add to db: %w", err)
		}

		if err = i.Enforcer.AddKnotMember(record.Domain, memberId.DID.String()); err != nil {
			return fmt.Errorf("failed to update ACLs: %w", err)
		}

		if err = tx.Commit(); err != nil {
			return fmt.Errorf("failed to commit txn: %w", err)
		}

		if err = i.Enforcer.E.SavePolicy(); err != nil {
			return fmt.Errorf("failed to save ACLs: %w", err)
		}
		committed = true

		l.Info("upserted knot member")
	case jmodels.CommitOperationDelete:
		rkey := e.Commit.RKey

		members, err := db.GetKnotMembers(
			i.Db,
			orm.FilterEq("did", did),
			orm.FilterEq("rkey", rkey),
		)
		if err != nil {
			return fmt.Errorf("failed to look up knot member with rkey %s: %w", rkey, err)
		}
		if len(members) == 0 {
			l.Info("knot member already removed", "rkey", rkey)
			return nil
		}
		if len(members) > 1 {
			return fmt.Errorf("multiple knot members with rkey %s", rkey)
		}
		member := members[0]

		tx, err := i.Db.Begin()
		if err != nil {
			return fmt.Errorf("failed to start txn: %w", err)
		}
		committed := false
		defer func() {
			if committed {
				return
			}
			tx.Rollback()
			i.Enforcer.E.LoadPolicy()
		}()

		if err = db.RemoveKnotMember(
			tx,
			orm.FilterEq("did", did),
			orm.FilterEq("rkey", rkey),
		); err != nil {
			return fmt.Errorf("failed to remove from db: %w", err)
		}

		if err = i.Enforcer.RemoveKnotMember(member.Domain, member.Subject.String()); err != nil {
			return fmt.Errorf("failed to update ACLs: %w", err)
		}

		if err = tx.Commit(); err != nil {
			return fmt.Errorf("failed to commit txn: %w", err)
		}

		if err = i.Enforcer.E.SavePolicy(); err != nil {
			return fmt.Errorf("failed to save ACLs: %w", err)
		}
		committed = true

		l.Info("removed knot member")
	}

	return nil
}

func (i *Ingester) ingestKnot(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	var err error

	l = l.With("handler", "ingestKnot")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.Knot{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		domain := e.Commit.RKey

		err := db.AddKnot(i.Db, domain, did)
		if err != nil {
			l.Error("failed to add knot to db", "err", err, "domain", domain)
			return err
		}

		if err := i.verifyKnot(ctx, domain, did); err != nil {
			l.Warn("failed to verify knot", "domain", domain, "did", did, "err", err)
		}

		l.Info("ingested record", "domain", domain)
		return nil

	case jmodels.CommitOperationDelete:
		domain := e.Commit.RKey

		// get record from db first
		registrations, err := db.GetRegistrations(
			i.Db,
			orm.FilterEq("domain", domain),
			orm.FilterEq("did", did),
		)
		if err != nil {
			return fmt.Errorf("failed to get registration: %w", err)
		}
		if len(registrations) != 1 {
			return fmt.Errorf("got incorrect number of registrations: %d, expected 1", len(registrations))
		}
		registration := registrations[0]

		tx, err := i.Db.Begin()
		if err != nil {
			return fmt.Errorf("failed to start txn: %w", err)
		}
		defer func() {
			tx.Rollback()
			i.Enforcer.E.LoadPolicy()
		}()

		err = db.RemoveKnotMember(
			tx,
			orm.FilterEq("did", did),
			orm.FilterEq("domain", domain),
		)
		if err != nil {
			return fmt.Errorf("failed to remove knot members: %w", err)
		}

		err = db.DeleteKnot(
			tx,
			orm.FilterEq("did", did),
			orm.FilterEq("domain", domain),
		)
		if err != nil {
			return fmt.Errorf("failed to delete knot: %w", err)
		}

		err = db.RemoveReposByKnot(tx, domain)
		if err != nil {
			return fmt.Errorf("failed to remove repos by knot: %w", err)
		}

		if registration.Registered != nil {
			err = i.Enforcer.RemoveKnot(domain)
			if err != nil {
				return fmt.Errorf("failed to remove knot from enforcer: %w", err)
			}
		}

		err = tx.Commit()
		if err != nil {
			return fmt.Errorf("failed to commit txn: %w", err)
		}

		err = i.Enforcer.E.SavePolicy()
		if err != nil {
			return fmt.Errorf("failed to save ACLs: %w", err)
		}

		l.Info("ingested record", "domain", domain)
	}

	return nil
}

const (
	verifyAttempts = 4
	verifyMinDelay = 1 * time.Second
	verifyMaxDelay = 5 * time.Second
)

func (i *Ingester) verifyKnot(ctx context.Context, domain, did string) error {
	regs, err := db.GetRegistrations(i.Db,
		orm.FilterEq("domain", domain),
		orm.FilterEq("did", did),
	)
	if err != nil {
		return fmt.Errorf("look up registration: %w", err)
	}
	if len(regs) != 1 {
		return fmt.Errorf("no registration for %s by %s", domain, did)
	}
	if regs[0].Registered != nil {
		return nil
	}

	err = retry.Do(
		func() error { return serververify.RunVerification(ctx, domain, did, i.Config.Core.Dev) },
		retry.Context(ctx),
		retry.Attempts(verifyAttempts),
		retry.Delay(verifyMinDelay),
		retry.MaxDelay(verifyMaxDelay),
		retry.DelayType(retry.BackOffDelay),
		retry.LastErrorOnly(true),
	)
	if err != nil {
		return fmt.Errorf("verify: %w", err)
	}
	return serververify.MarkKnotVerified(i.Db, i.Enforcer, domain, did)
}

func (i *Ingester) verifySpindle(ctx context.Context, instance, did string) error {
	spindles, err := db.GetSpindles(ctx, i.Db,
		orm.FilterEq("instance", instance),
		orm.FilterEq("owner", did),
	)
	if err != nil {
		return fmt.Errorf("look up spindle: %w", err)
	}
	if len(spindles) != 1 {
		return fmt.Errorf("no spindle for %s by %s", instance, did)
	}
	if spindles[0].Verified != nil {
		return nil
	}

	err = retry.Do(
		func() error { return serververify.RunVerification(ctx, instance, did, i.Config.Core.Dev) },
		retry.Context(ctx),
		retry.Attempts(verifyAttempts),
		retry.Delay(verifyMinDelay),
		retry.MaxDelay(verifyMaxDelay),
		retry.DelayType(retry.BackOffDelay),
		retry.LastErrorOnly(true),
	)
	if err != nil {
		return fmt.Errorf("verify: %w", err)
	}
	_, err = serververify.MarkSpindleVerified(i.Db, i.Enforcer, instance, did)
	return err
}

const sweepConcurrency = 4

func (i *Ingester) SweepPendingVerifications() {
	l := i.Logger.With("handler", "SweepPendingVerifications")

	var g errgroup.Group
	g.SetLimit(sweepConcurrency)

	regs, err := db.GetRegistrations(i.Db, orm.FilterIs("registered", nil))
	if err != nil {
		l.Error("failed to list unverified knots", "err", err)
	} else {
		for _, reg := range regs {
			g.Go(func() error {
				if err := i.verifyKnot(i.Ctx, reg.Domain, reg.ByDid); err != nil {
					l.Warn("verify knot failed", "domain", reg.Domain, "did", reg.ByDid, "err", err)
				}
				return nil
			})
		}
	}

	spindles, err := db.GetSpindles(i.Ctx, i.Db, orm.FilterIs("verified", nil))
	if err != nil {
		l.Error("failed to list unverified spindles", "err", err)
		g.Wait()
		return
	}
	for _, s := range spindles {
		g.Go(func() error {
			if err := i.verifySpindle(i.Ctx, s.Instance, s.Owner.String()); err != nil {
				l.Warn("verify spindle failed", "instance", s.Instance, "owner", s.Owner, "err", err)
			}
			return nil
		})
	}
	g.Wait()
}

func (i *Ingester) ingestIssue(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	rkey := e.Commit.RKey

	var err error

	l = l.With("handler", "ingestIssue")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.RepoIssue{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		issue := models.IssueFromRecord(did, rkey, record)

		if issue.RepoDid == "" {
			return fmt.Errorf("issue record has no repo field")
		}
		if _, err := syntax.ParseDID(string(issue.RepoDid)); err != nil {
			return fmt.Errorf("issue record repo field is not a valid DID: %w", err)
		}

		if err := issue.Validate(); err != nil {
			return fmt.Errorf("failed to validate issue: %w", err)
		}

		if record.Repo != "" && !strings.HasPrefix(record.Repo, "did:") {
			repo, repoErr := db.GetRepoByAtUri(i.Db, record.Repo)
			if repoErr == nil && repo.RepoDid != "" {
				if enqErr := db.EnqueuePdsRecordMigration(ctx, i.Db, "add-repo-did", syntax.DID(did), syntax.NSID(tangled.RepoIssueNSID), syntax.RecordKey(e.Commit.RKey)); enqErr != nil {
					l.Warn("failed to enqueue PDS rewrite for issue", "err", enqErr, "did", did, "repoDid", repo.RepoDid)
				}
			}
		}

		tx, err := i.Db.BeginTx(ctx, nil)
		if err != nil {
			l.Error("failed to begin transaction", "err", err)
			return err
		}
		defer tx.Rollback()

		err = db.PutIssue(tx, &issue)
		if err != nil {
			l.Error("failed to create issue", "err", err)
			return err
		}

		if err := db.ResolveIssueState(tx, issue.AtUri()); err != nil {
			l.Error("failed to resolve issue state", "err", err)
			return err
		}

		err = tx.Commit()
		if err != nil {
			l.Error("failed to commit txn", "err", err)
			return err
		}

		i.drainPendingState(ctx, issue.AtUri(), issueStateSpec, l)

		l.Info("ingested record")
		return nil

	case jmodels.CommitOperationDelete:
		tx, err := i.Db.BeginTx(ctx, nil)
		if err != nil {
			l.Error("failed to begin transaction", "err", err)
			return err
		}
		defer tx.Rollback()

		if err := db.DeleteIssues(
			tx,
			did,
			rkey,
		); err != nil {
			l.Error("failed to delete", "err", err)
			return fmt.Errorf("failed to delete issue record: %w", err)
		}
		if err := tx.Commit(); err != nil {
			l.Error("failed to commit txn", "err", err)
			return err
		}

		l.Info("ingested record")
		return nil
	}

	return nil
}

func (i *Ingester) ingestPull(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	rkey := e.Commit.RKey

	var err error

	l = l.With("handler", "ingestPull")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.RepoPull{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			l.Error("invalid record", "err", err)
			return err
		}

		ownerId, err := i.IdResolver.ResolveIdent(ctx, did)
		if err != nil {
			l.Error("failed to resolve did", "err", err)
			return err
		}

		// go through and fetch all blobs in parallel
		readers := make([]*io.ReadCloser, len(record.Rounds))
		var mu sync.Mutex

		g, gctx := errgroup.WithContext(ctx)

		for idx, b := range record.Rounds {
			g.Go(func() error {
				// for some reason, a blob is empty
				if b.PatchBlob == nil {
					return fmt.Errorf("missing patchBlob in round %d", idx)
				}

				ownerPds := ownerId.PDSEndpoint()
				url, _ := url.Parse(fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob", ownerPds))
				q := url.Query()
				q.Set("cid", b.PatchBlob.Ref.String())
				q.Set("did", did)
				url.RawQuery = q.Encode()

				req, err := http.NewRequestWithContext(gctx, http.MethodGet, url.String(), nil)
				if err != nil {
					l.Error("failed to create request")
					return err
				}
				req.Header.Set("Content-Type", "application/json")

				resp, err := http.DefaultClient.Do(req)
				if err != nil {
					l.Error("failed to make request")
					return err
				}

				mu.Lock()
				readers[idx] = &resp.Body
				mu.Unlock()

				return nil
			})
		}

		if err := g.Wait(); err != nil {
			for _, r := range readers {
				if r != nil && *r != nil {
					(*r).Close()
				}
			}
			return err
		}

		defer func() {
			for _, r := range readers {
				if r != nil && *r != nil {
					(*r).Close()
				}
			}
		}()

		pull, err := models.PullFromRecord(did, rkey, record, readers)
		if err != nil {
			return fmt.Errorf("failed to parse pull from record: %w", err)
		}
		if err := pull.Validate(); err != nil {
			return fmt.Errorf("failed to validate pull: %w", err)
		}
		if pull.DependentOn != nil {
			if err := func() error {
				dependentPull, err := db.GetPull(
					i.Db,
					orm.FilterEq("dependent_on", pull.DependentOn.String()),
				)
				if errors.Is(err, sql.ErrNoRows) {
					return nil
				}
				if err != nil {
					return fmt.Errorf("failed to fetch pulls with same dependency: %w", err)
				}
				if dependentPull.AtUri() == pull.AtUri() {
					return nil
				}
				return fmt.Errorf("another pull already depends on %s, which would form a DAG, this is presently disallowed", pull.DependentOn.String())
			}(); err != nil {
				return fmt.Errorf("failed to validate pull stack: %w", err)
			}
		}

		tx, err := i.Db.BeginTx(ctx, nil)
		if err != nil {
			l.Error("failed to begin transaction", "err", err)
			return err
		}
		defer tx.Rollback()

		err = db.PutPull(tx, pull)
		if err != nil {
			l.Error("failed to create pull", "err", err)
			return err
		}

		if err := db.ResolvePullStatus(tx, pull.AtUri()); err != nil {
			l.Error("failed to resolve pull status", "err", err)
			return err
		}

		err = tx.Commit()
		if err != nil {
			l.Error("failed to commit txn", "err", err)
			return err
		}

		i.drainPendingState(ctx, pull.AtUri(), pullStatusSpec, l)

		l.Info("ingested record")
		return nil

	case jmodels.CommitOperationDelete:
		tx, err := i.Db.BeginTx(ctx, nil)
		if err != nil {
			l.Error("failed to begin transaction", "err", err)
			return err
		}
		defer tx.Rollback()

		if err := db.AbandonPulls(
			tx,
			orm.FilterEq("owner_did", did),
			orm.FilterEq("rkey", rkey),
		); err != nil {
			l.Error("failed to abandon", "err", err)
			return fmt.Errorf("failed to abandon pull record: %w", err)
		}
		if err := tx.Commit(); err != nil {
			l.Error("failed to commit txn", "err", err)
			return err
		}

		l.Info("ingested record")
		return nil
	}

	return nil
}

func (i *Ingester) authorizeStateRecord(ctx context.Context, repo *models.Repo, subjectAuthorDid, recordAuthorDid string, l *slog.Logger) (bool, error) {
	if recordAuthorDid == subjectAuthorDid {
		return true, nil
	}
	if recordAuthorDid == consts.TangledDid {
		return true, nil
	}

	ok, err := i.Acl.HasRepoPermissionErr(ctx, repo, recordAuthorDid, "repo:push")
	if err != nil {
		if errors.Is(err, knotacl.ErrKnotUnreachable) {
			l.Warn("ingesting state record without permission check", "did", recordAuthorDid, "err", err)
			return true, nil
		}
		return false, err
	}
	return ok, nil
}

type stateIngestSpec struct {
	subjectNSID string
	parse       func(did, rkey string, raw json.RawMessage) (models.StateRecord, error)
	findSubject func(e db.Execer, subject syntax.ATURI) (repo *models.Repo, authorDid string, found bool, err error)
	put         func(tx *sql.Tx, rec models.StateRecord) (syntax.ATURI, error)
	resolve     func(tx *sql.Tx, subject syntax.ATURI) error
	recompute   func(tx *sql.Tx, subject syntax.ATURI) error
	del         func(tx *sql.Tx, did, rkey string) (syntax.ATURI, error)
}

var issueStateSpec = stateIngestSpec{
	subjectNSID: tangled.RepoIssueNSID,
	parse: func(did, rkey string, raw json.RawMessage) (models.StateRecord, error) {
		record := tangled.RepoIssueState{}
		if err := json.Unmarshal(raw, &record); err != nil {
			return models.StateRecord{}, fmt.Errorf("invalid record: %w", err)
		}
		return models.IssueStateFromRecord(did, rkey, record)
	},
	findSubject: func(e db.Execer, subject syntax.ATURI) (*models.Repo, string, bool, error) {
		issues, err := db.GetIssues(e, orm.FilterEq("at_uri", subject))
		if err != nil {
			return nil, "", false, err
		}
		if len(issues) != 1 || issues[0].Repo == nil {
			return nil, "", false, nil
		}
		return issues[0].Repo, issues[0].Did, true, nil
	},
	put:       db.PutIssueState,
	resolve:   db.ResolveIssueState,
	recompute: db.RecomputeIssueState,
	del:       db.DeleteIssueState,
}

var pullStatusSpec = stateIngestSpec{
	subjectNSID: tangled.RepoPullNSID,
	parse: func(did, rkey string, raw json.RawMessage) (models.StateRecord, error) {
		record := tangled.RepoPullStatus{}
		if err := json.Unmarshal(raw, &record); err != nil {
			return models.StateRecord{}, fmt.Errorf("invalid record: %w", err)
		}
		return models.PullStatusFromRecord(did, rkey, record)
	},
	findSubject: func(e db.Execer, subject syntax.ATURI) (*models.Repo, string, bool, error) {
		pulls, err := db.GetPulls(e, orm.FilterEq("at_uri", subject))
		if err != nil {
			return nil, "", false, err
		}
		if len(pulls) != 1 || pulls[0].Repo == nil {
			return nil, "", false, nil
		}
		return pulls[0].Repo, pulls[0].OwnerDid, true, nil
	},
	put:       db.PutPullStatus,
	resolve:   db.ResolvePullStatus,
	recompute: db.RecomputePullStatus,
	del:       db.DeletePullStatus,
}

func (i *Ingester) ingestState(ctx context.Context, e *jmodels.Event, l *slog.Logger, spec stateIngestSpec) error {
	did := e.Did
	rkey := e.Commit.RKey
	nsid := e.Commit.Collection

	l = l.With("handler", "ingestState", "nsid", nsid)

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		return i.applyStateRecord(ctx, did, rkey, nsid, e.Commit.Record, spec, l)
	case jmodels.CommitOperationDelete:
		return i.deleteStateRecord(ctx, did, rkey, nsid, spec, l)
	}

	return nil
}

func (i *Ingester) applyStateRecord(ctx context.Context, did, rkey, nsid string, raw []byte, spec stateIngestSpec, l *slog.Logger) error {
	rec, err := spec.parse(did, rkey, json.RawMessage(raw))
	if err != nil {
		return err
	}
	if string(rec.Subject.Collection()) != spec.subjectNSID {
		return fmt.Errorf("state subject is not %s: %s", spec.subjectNSID, rec.Subject)
	}

	repo, authorDid, found, err := spec.findSubject(i.Db, rec.Subject)
	if err != nil {
		return fmt.Errorf("failed to look up state subject: %w", err)
	}
	if !found {
		return i.parkStateRecord(ctx, did, rkey, nsid, rec.Subject, raw, l)
	}

	authorized, err := i.authorizeStateRecord(ctx, repo, authorDid, did, l)
	if err != nil {
		return i.parkStateRecord(ctx, did, rkey, nsid, rec.Subject, raw, l)
	}

	tx, err := i.Db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	if err := db.UnparkStateRecord(tx, did, rkey, nsid); err != nil {
		return fmt.Errorf("failed to unpark state record: %w", err)
	}

	if !authorized {
		if err := tx.Commit(); err != nil {
			return err
		}
		l.Warn("dropped unauthorized state record", "did", did, "rkey", rkey, "subject", rec.Subject)
		return nil
	}

	priorSubject, err := spec.put(tx, rec)
	if err != nil {
		return fmt.Errorf("failed to put state record: %w", err)
	}
	if err := spec.resolve(tx, rec.Subject); err != nil {
		return fmt.Errorf("failed to resolve state: %w", err)
	}
	if priorSubject != "" {
		if err := spec.recompute(tx, priorSubject); err != nil {
			return fmt.Errorf("failed to recompute prior subject state: %w", err)
		}
	}

	if err := tx.Commit(); err != nil {
		return err
	}

	l.Info("ingested record")
	return nil
}

func (i *Ingester) deleteStateRecord(ctx context.Context, did, rkey, nsid string, spec stateIngestSpec, l *slog.Logger) error {
	tx, err := i.Db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	if err := db.UnparkStateRecord(tx, did, rkey, nsid); err != nil {
		return fmt.Errorf("failed to unpark state record: %w", err)
	}
	subject, err := spec.del(tx, did, rkey)
	if err != nil {
		return fmt.Errorf("failed to delete state record: %w", err)
	}
	if subject != "" {
		if err := spec.recompute(tx, subject); err != nil {
			return fmt.Errorf("failed to recompute state: %w", err)
		}
	}

	if err := tx.Commit(); err != nil {
		return err
	}

	l.Info("ingested record")
	return nil
}

func (i *Ingester) parkStateRecord(ctx context.Context, did, rkey, nsid string, subject syntax.ATURI, raw []byte, l *slog.Logger) error {
	tx, err := i.Db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	if err := db.ParkStateRecord(tx, db.PendingStateRecord{
		Did:     did,
		Rkey:    rkey,
		Nsid:    nsid,
		Subject: subject,
		Record:  raw,
	}); err != nil {
		return fmt.Errorf("failed to park state record: %w", err)
	}

	if err := tx.Commit(); err != nil {
		return err
	}

	l.Info("parked state record for retry", "subject", subject)
	return nil
}

func (i *Ingester) drainPendingState(ctx context.Context, subject syntax.ATURI, spec stateIngestSpec, l *slog.Logger) {
	pending, err := db.PendingStateRecordsForSubject(i.Db, subject)
	if err != nil {
		l.Error("failed to load pending state records", "err", err, "subject", subject)
		return
	}
	for _, p := range pending {
		if err := i.applyStateRecord(ctx, p.Did, p.Rkey, p.Nsid, p.Record, spec, l); err != nil {
			l.Error("failed to drain pending state record", "err", err, "did", p.Did, "rkey", p.Rkey)
		}
	}
}

const (
	pendingStateReconcileInterval = time.Hour
	pendingStateRecordTTL         = 7 * 24 * time.Hour
)

func stateSpecForSubject(subject syntax.ATURI) (stateIngestSpec, bool) {
	switch string(subject.Collection()) {
	case tangled.RepoIssueNSID:
		return issueStateSpec, true
	case tangled.RepoPullNSID:
		return pullStatusSpec, true
	default:
		return stateIngestSpec{}, false
	}
}

func (i *Ingester) StartPendingStateReconciler() {
	i.ReconcilePendingState()

	ticker := time.NewTicker(pendingStateReconcileInterval)
	defer ticker.Stop()
	for {
		select {
		case <-i.Ctx.Done():
			return
		case <-ticker.C:
			i.ReconcilePendingState()
		}
	}
}

func (i *Ingester) ReconcilePendingState() {
	l := i.Logger.With("handler", "reconcilePendingState")

	subjects, err := db.DistinctPendingStateSubjects(i.Db)
	if err != nil {
		l.Error("failed to list pending state subjects", "err", err)
	}
	for _, subject := range subjects {
		spec, ok := stateSpecForSubject(subject)
		if !ok {
			continue
		}
		i.drainPendingState(i.Ctx, subject, spec, l)
	}

	cutoff := time.Now().Add(-pendingStateRecordTTL).UTC().Format(time.RFC3339)
	evicted, err := db.EvictStalePendingStateRecords(i.Db, cutoff)
	if err != nil {
		l.Error("failed to evict stale pending state records", "err", err)
		return
	}
	if evicted > 0 {
		l.Warn("evicted stale pending state records", "count", evicted, "olderThan", cutoff)
	}
}

// ingestIssueComment ingests legacy sh.tangled.repo.issue.comment deletions
func (i *Ingester) ingestIssueComment(e *jmodels.Event, l *slog.Logger) error {
	l = l.With("handler", "ingestIssueComment")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		// no-op. sh.tangled.repo.issue.comment is deprecated

	case jmodels.CommitOperationDelete:
		if err := db.PurgeComments(
			i.Db,
			orm.FilterEq("did", e.Did),
			orm.FilterEq("collection", e.Commit.Collection),
			orm.FilterEq("rkey", e.Commit.RKey),
		); err != nil {
			return fmt.Errorf("failed to delete comment record: %w", err)
		}
	}

	l.Info("ingested record")
	return nil
}

// ingestPullComment ingests legacy sh.tangled.repo.pull.comment deletions
func (i *Ingester) ingestPullComment(e *jmodels.Event, l *slog.Logger) error {
	l = l.With("handler", "ingestPullComment")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		// no-op. sh.tangled.repo.pull.comment is deprecated

	case jmodels.CommitOperationDelete:
		if err := db.PurgeComments(
			i.Db,
			orm.FilterEq("did", e.Did),
			orm.FilterEq("collection", e.Commit.Collection),
			orm.FilterEq("rkey", e.Commit.RKey),
		); err != nil {
			return fmt.Errorf("failed to delete comment record: %w", err)
		}
	}

	l.Info("ingested record")
	return nil
}

func (i *Ingester) ingestComment(e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	rkey := e.Commit.RKey
	cid := e.Commit.CID

	var err error

	l = l.With("handler", "ingestComment")

	ctx := context.Background()

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.FeedComment{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			return fmt.Errorf("invalid record: %w", err)
		}

		comment, err := models.CommentFromRecord(syntax.DID(did), syntax.RecordKey(rkey), syntax.CID(cid), record)
		if err != nil {
			return fmt.Errorf("failed to parse comment from record: %w", err)
		}

		if err := comment.Validate(); err != nil {
			return fmt.Errorf("failed to validate comment: %w", err)
		}

		var references []syntax.ATURI
		if comment.Body.Original != nil {
			_, references = i.MentionsResolver.Resolve(ctx, *comment.Body.Original)
		}

		tx, err := i.Db.Begin()
		if err != nil {
			return fmt.Errorf("failed to start transaction: %w", err)
		}
		defer tx.Rollback()

		_, err = db.PutComment(tx, comment, references)
		if err != nil {
			return fmt.Errorf("failed to create comment: %w", err)
		}

		if err := tx.Commit(); err != nil {
			return err
		}

	case jmodels.CommitOperationDelete:
		if err := db.DeleteComments(
			i.Db,
			orm.FilterEq("did", did),
			orm.FilterEq("collection", e.Commit.Collection),
			orm.FilterEq("rkey", rkey),
		); err != nil {
			return fmt.Errorf("failed to delete comment record: %w", err)
		}
	}

	l.Info("ingested record")
	return nil
}

func (i *Ingester) ingestReaction(e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	rkey := e.Commit.RKey

	l = l.With("handler", "ingestReaction")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.FeedReaction{}
		if err := json.Unmarshal(raw, &record); err != nil {
			return fmt.Errorf("invalid record: %w", err)
		}

		subjectUri, err := syntax.ParseATURI(record.Subject)
		if err != nil {
			return fmt.Errorf("invalid reaction subject %q: %w", record.Subject, err)
		}
		subjectUri = models.NormalizeReactionSubject(subjectUri)

		kind, ok := models.ParseReactionKind(record.Reaction)
		if !ok {
			return fmt.Errorf("invalid reaction kind: %q", record.Reaction)
		}

		created, parseErr := time.Parse(time.RFC3339, record.CreatedAt)
		if parseErr != nil {
			created = time.Now()
		}

		reaction := models.Reaction{
			ReactedByDid: did,
			Rkey:         rkey,
			ThreadAt:     subjectUri,
			Kind:         kind,
			Created:      created,
		}
		if err := db.UpsertReaction(i.Db, reaction); err != nil {
			return fmt.Errorf("failed to upsert reaction: %w", err)
		}

	case jmodels.CommitOperationDelete:
		if err := db.DeleteReactionByRkey(i.Db, did, rkey); err != nil {
			return fmt.Errorf("failed to delete reaction record: %w", err)
		}
	}

	l.Info("ingested record")
	return nil
}

func (i *Ingester) ingestLabelDefinition(e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	rkey := e.Commit.RKey

	var err error

	l = l.With("handler", "ingestLabelDefinition")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.LabelDefinition{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			return fmt.Errorf("invalid record: %w", err)
		}

		def, err := models.LabelDefinitionFromRecord(did, rkey, record)
		if err != nil {
			return fmt.Errorf("failed to parse labeldef from record: %w", err)
		}

		if err := def.Validate(); err != nil {
			return fmt.Errorf("failed to validate labeldef: %w", err)
		}

		_, err = db.AddLabelDefinition(i.Db, def)
		if err != nil {
			return fmt.Errorf("failed to create labeldef: %w", err)
		}

		l.Info("ingested record")
		return nil

	case jmodels.CommitOperationDelete:
		if err := db.DeleteLabelDefinition(
			i.Db,
			orm.FilterEq("did", did),
			orm.FilterEq("rkey", rkey),
		); err != nil {
			return fmt.Errorf("failed to delete labeldef record: %w", err)
		}

		l.Info("ingested record")
		return nil
	}

	return nil
}

func (i *Ingester) ingestLabelOp(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {
	did := e.Did
	rkey := e.Commit.RKey

	var err error

	l = l.With("handler", "ingestLabelOp")

	switch e.Commit.Operation {
	case jmodels.CommitOperationCreate:
		raw := json.RawMessage(e.Commit.Record)
		record := tangled.LabelOp{}
		err = json.Unmarshal(raw, &record)
		if err != nil {
			return fmt.Errorf("invalid record: %w", err)
		}

		subject := syntax.ATURI(record.Subject)
		collection := subject.Collection()

		var repo *models.Repo
		switch collection {
		case tangled.RepoIssueNSID:
			i, err := db.GetIssues(i.Db, orm.FilterEq("at_uri", subject))
			if err != nil || len(i) != 1 {
				return fmt.Errorf("failed to find subject: %w || subject count %d", err, len(i))
			}
			repo = i[0].Repo
		case tangled.RepoPullNSID:
			p, err := db.GetPulls(i.Db, orm.FilterEq("at_uri", subject))
			if err != nil || len(p) != 1 {
				return fmt.Errorf("failed to find subject: %w || subject count %d", err, len(p))
			}
			repo = p[0].Repo
		default:
			return fmt.Errorf("unsupported label subject: %s", collection)
		}

		actx, err := db.NewLabelApplicationCtx(i.Db, orm.FilterIn("at_uri", repo.Labels))
		if err != nil {
			return fmt.Errorf("failed to build label application ctx: %w", err)
		}

		ops := models.LabelOpsFromRecord(did, rkey, record)

		for _, o := range ops {
			def, ok := actx.Defs[o.OperandKey]
			if !ok {
				return fmt.Errorf("failed to find label def for key: %s, expected: %q", o.OperandKey, slices.Collect(maps.Keys(actx.Defs)))
			}
			// validate permissions: only collaborators can apply labels currently
			//
			// TODO: introduce a repo:triage permission
			allowed, permErr := i.Acl.HasRepoPermissionErr(ctx, repo, o.Did, "repo:push")
			if permErr != nil {
				if !errors.Is(permErr, knotacl.ErrKnotUnreachable) {
					return fmt.Errorf("enforcing permission: %w", permErr)
				}
				l.Warn("ingesting labelop without permission check", "did", o.Did, "err", permErr)
			} else if !allowed {
				return fmt.Errorf("unauthorized label operation")
			}

			if err := def.ValidateOperandValue(&o); err != nil {
				return fmt.Errorf("failed to validate labelop: %w", err)
			}
		}

		tx, err := i.Db.Begin()
		if err != nil {
			return err
		}
		defer tx.Rollback()

		for _, o := range ops {
			_, err = db.AddLabelOp(tx, &o)
			if err != nil {
				return fmt.Errorf("failed to add labelop: %w", err)
			}
		}

		if err = tx.Commit(); err != nil {
			return err
		}

		l.Info("ingested record")
	}

	return nil
}