diff --git a/pkg/atproto/atproto.go b/pkg/atproto/atproto.go
index e5799aa1..82aaced5 100644
--- a/pkg/atproto/atproto.go
+++ b/pkg/atproto/atproto.go
@@ -30,6 +30,14 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepoCached(ctx context.Context, ha
 	if err != nil {
 		return nil, fmt.Errorf("failed to get repo for %s: %w", handle, err)
 	}
+	// A terminal account status -- deactivated, deleted, taken down -- means
+	// syncing is pointless until the account comes back, which the firehose
+	// tells us about by clearing the status. Hand back what we have rather than
+	// asking a PDS a question we know the answer to.
+	if repo != nil && repo.TerminalStatus() {
+		log.Debug(ctx, "skipping sync of repo in terminal state", "did", repo.DID, "status", repo.Status)
+		return repo, nil
+	}
 	// An empty Version means the row is a placeholder written at the start of a
 	// backfill that never finished, so the repo is only partially indexed. Fall
 	// through and sync it again -- unless the backfill that wrote it is still
@@ -103,15 +111,21 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle s
 
 	rev, rootCID, err := atsync.backfillRepo(ctx, ident, &xrpcc)
 	if err != nil {
+		if parked := parkTerminalRepo(ctx, mod, ident.DID.String(), err); parked != nil {
+			return nil, parked
+		}
 		return nil, err
 	}
 
+	// A completed backfill proves the account is fine, so Status goes back to
+	// empty -- UpdateRepo writes every column, so this happens by construction.
 	newRepo := model.Repo{
 		DID:     ident.DID.String(),
 		PDS:     ident.PDSEndpoint(),
 		Version: rev,
 		RootCID: rootCID,
 		Handle:  ident.Handle.String(),
+		Status:  model.RepoStatusOK,
 	}
 	err = mod.UpdateRepo(&newRepo)
 	if err != nil {
@@ -177,6 +191,10 @@ func (atsync *ATProtoSynchronizer) RefreshIdentity(ctx context.Context, did stri
 	if oldRepo != nil {
 		newRepo.Version = oldRepo.Version
 		newRepo.RootCID = oldRepo.RootCID
+		// Same reasoning for Status: an identity event is not evidence the
+		// account came back, and blanking it here would put every deactivated
+		// repo back in the boot-time sync sweep.
+		newRepo.Status = oldRepo.Status
 	}
 	err = atsync.Model.UpdateRepo(&newRepo)
 	if err != nil {
diff --git a/pkg/atproto/backfill_walk.go b/pkg/atproto/backfill_walk.go
index 8eb42160..e49eb77f 100644
--- a/pkg/atproto/backfill_walk.go
+++ b/pkg/atproto/backfill_walk.go
@@ -16,6 +16,7 @@ import (
 	"github.com/bluesky-social/indigo/xrpc"
 	"github.com/ipfs/go-cid"
 	"stream.place/streamplace/pkg/log"
+	"stream.place/streamplace/pkg/model"
 	"stream.place/streamplace/pkg/reposync"
 )
 
@@ -380,6 +381,63 @@ func isLexiconErrorName(s string) bool {
 	return true
 }
 
+// TerminalRepoError is a backfill failure that retrying cannot fix, because the
+// account itself is deactivated, deleted, suspended, or taken down. Callers
+// record Status on the repo row and stop; the firehose clears it if the account
+// comes back.
+type TerminalRepoError struct {
+	// Status is a model.RepoStatus* value.
+	Status string
+	Err    error
+}
+
+func (e *TerminalRepoError) Error() string {
+	return fmt.Sprintf("repo is %s: %v", e.Status, e.Err)
+}
+
+func (e *TerminalRepoError) Unwrap() error { return e.Err }
+
+// repoStatusFromError maps the account-lifecycle errors an atproto host returns
+// onto a [model.Repo] status, or "" for anything worth retrying.
+//
+// Only named lexicon errors count. A DNS failure, a timeout, an SSRF block or a
+// 500 says nothing about the account -- parking a repo as "notfound" because its
+// PDS was briefly unreachable would take it out of the index until it happened
+// to commit again.
+func repoStatusFromError(err error) string {
+	switch xrpcErrorName(err) {
+	case "RepoDeactivated":
+		return model.RepoStatusDeactivated
+	case "RepoNotFound":
+		return model.RepoStatusNotFound
+	case "RepoTakendown":
+		return model.RepoStatusTakendown
+	case "RepoSuspended":
+		return model.RepoStatusSuspended
+	}
+	return ""
+}
+
+// parkTerminalRepo records a terminal account state on the repo row so that the
+// boot-time sweep and every cached lookup stop asking about it.
+//
+// It returns the error the caller should return -- a [TerminalRepoError] if the
+// failure was terminal, or nil if it was not, in which case the caller keeps its
+// own error. Version and RootCID are deliberately left as they are: whatever we
+// managed to index of this repo before it went away stays valid and stays
+// served.
+func parkTerminalRepo(ctx context.Context, mod model.Model, did string, err error) error {
+	status := repoStatusFromError(err)
+	if status == model.RepoStatusOK {
+		return nil
+	}
+	log.Log(ctx, "repo is in a terminal account state", "did", did, "status", status, "err", err)
+	if serr := mod.SetRepoStatus(ctx, did, status); serr != nil {
+		return fmt.Errorf("failed to record %s status for %s: %w", status, did, serr)
+	}
+	return &TerminalRepoError{Status: status, Err: err}
+}
+
 // isMethodNotSupported reports whether err means "this host does not implement
 // that XRPC method", which is the only failure the backfill is allowed to
 // answer by falling back to a full getRepo.
diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go
index aed0aa0a..9179090d 100644
--- a/pkg/atproto/firehose.go
+++ b/pkg/atproto/firehose.go
@@ -472,6 +472,7 @@ func (atsync *ATProtoSynchronizer) handleCommitEventOps(ctx context.Context, evt
 			log.Error(ctx, "failed to get repo", "err", err)
 			continue
 		}
+		atsync.reviveRepo(ctx, r)
 		// log.Warn(ctx, "got record we care about", "collection", collection, "rkey", rkey)
 
 		ek := repomgr.EventKind(op.Action)
@@ -696,6 +697,25 @@ func (atsync *ATProtoSynchronizer) handleCommitEventOps(ctx context.Context, evt
 	}
 }
 
+// reviveRepo un-parks a repo we had written off. A commit event is proof the
+// account is back -- a deactivated, suspended, or deleted repo cannot write --
+// so the terminal status goes away and the ordinary wedge logic (an empty
+// Version means "backfill me") takes it from there.
+//
+// It is called from the commit path with the row that path already loaded, so
+// the common case costs one comparison and no query at all.
+func (atsync *ATProtoSynchronizer) reviveRepo(ctx context.Context, r *model.Repo) {
+	if !r.TerminalStatus() {
+		return
+	}
+	log.Log(ctx, "repo committed while parked, clearing terminal status", "did", r.DID, "status", r.Status)
+	if err := atsync.Model.SetRepoStatus(ctx, r.DID, model.RepoStatusOK); err != nil {
+		log.Error(ctx, "failed to clear repo status", "did", r.DID, "err", err)
+		return
+	}
+	r.Status = model.RepoStatusOK
+}
+
 func (atsync *ATProtoSynchronizer) handleIdentityEventOps(ctx context.Context, evt *indigoatproto.SyncSubscribeRepos_Identity) {
 	handle := ""
 	if evt.Handle != nil {
diff --git a/pkg/atproto/migrate.go b/pkg/atproto/migrate.go
index cff9a165..89df958e 100644
--- a/pkg/atproto/migrate.go
+++ b/pkg/atproto/migrate.go
@@ -12,7 +12,20 @@ import (
 )
 
 func (atsync *ATProtoSynchronizer) Migrate(ctx context.Context) error {
+	// Accounts that are deactivated, deleted, or taken down fail their backfill
+	// the same way on every boot forever. One query up front keeps them out of
+	// the sweep entirely, instead of one logged failure each.
+	terminalDIDs, err := atsync.Model.TerminalRepoDIDs(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to list repos in terminal states: %w", err)
+	}
+	terminal := make(map[string]struct{}, len(terminalDIDs))
+	for _, did := range terminalDIDs {
+		terminal[did] = struct{}{}
+	}
+
 	var allDIDs []string
+	skipped := 0
 	offset := 0
 	for {
 		repos, err := atsync.StatefulDB.ListRepos(100, offset)
@@ -23,11 +36,18 @@ func (atsync *ATProtoSynchronizer) Migrate(ctx context.Context) error {
 			break
 		}
 		for _, repo := range repos {
+			if _, ok := terminal[repo.DID]; ok {
+				skipped++
+				continue
+			}
 			allDIDs = append(allDIDs, repo.DID)
 		}
 		offset += len(repos)
 	}
 
+	if skipped > 0 {
+		log.Log(ctx, "skipping repos with terminal status", "skipped", skipped)
+	}
 	log.Log(ctx, "starting migration sync", "totalRepos", len(allDIDs))
 
 	g, ctx := errgroup.WithContext(ctx)
diff --git a/pkg/atproto/redelivery_test.go b/pkg/atproto/redelivery_test.go
new file mode 100644
index 00000000..fb639fa8
--- /dev/null
+++ b/pkg/atproto/redelivery_test.go
@@ -0,0 +1,270 @@
+package atproto
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/bluesky-social/indigo/atproto/syntax"
+	"github.com/bluesky-social/indigo/util"
+	"github.com/bluesky-social/indigo/xrpc"
+	"github.com/stretchr/testify/require"
+	"stream.place/streamplace/pkg/bus"
+	"stream.place/streamplace/pkg/config"
+	"stream.place/streamplace/pkg/model"
+	"stream.place/streamplace/pkg/placestream"
+	"stream.place/streamplace/pkg/spid"
+	"stream.place/streamplace/pkg/statedb"
+)
+
+// offlineSynchronizer builds a synchronizer with no network anywhere: the PLC
+// URL points at a port nothing listens on, so any test that accidentally
+// resolves an identity fails fast and loudly instead of reaching the internet.
+func offlineSynchronizer(t *testing.T) (*ATProtoSynchronizer, model.Model, *bus.Bus) {
+	t.Helper()
+	cli := config.CLI{
+		BroadcasterHost: "example.com",
+		DBURL:           ":memory:",
+		DataDir:         t.TempDir(),
+		PLCURL:          "http://127.0.0.1:1",
+	}
+	mod, err := model.MakeDB(":memory:")
+	require.NoError(t, err)
+	state, err := statedb.MakeDB(context.Background(), &cli, nil, mod)
+	require.NoError(t, err)
+	b := bus.NewBus()
+	return &ATProtoSynchronizer{
+		CLI:        &cli,
+		StatefulDB: state,
+		Model:      mod,
+		Bus:        b,
+	}, mod, b
+}
+
+// TestHandleCreateUpdateRedelivery is the whole point of the idempotent
+// indexer: the same chat message delivered twice -- a cursor replay, a re-walk,
+// two relays carrying the same commit -- must be indexed once and must reach
+// the chat bus once. A second fanout would show the message twice in every
+// viewer's chat.
+func TestHandleCreateUpdateRedelivery(t *testing.T) {
+	ctx := context.Background()
+	atsync, mod, b := offlineSynchronizer(t)
+
+	did := "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa"
+	// A finished repo row: SyncBlueskyRepoCached short-circuits on it, so
+	// indexing never goes near the network.
+	require.NoError(t, mod.UpdateRepo(&model.Repo{
+		DID:     did,
+		Handle:  "chatter.test",
+		PDS:     "http://127.0.0.1:1",
+		Version: "3lrev00000000",
+	}))
+
+	rec := &placestream.ChatMessage{
+		LexiconTypeID: "place.stream.chat.message",
+		Text:          "hello twice",
+		CreatedAt:     time.Now().UTC().Format(util.ISO8601),
+		Streamer:      did,
+	}
+	var buf bytes.Buffer
+	require.NoError(t, rec.MarshalCBOR(&buf))
+	recCBOR := buf.Bytes()
+	rcid, err := spid.GetCID(rec)
+	require.NoError(t, err)
+
+	ch := b.Subscribe(did)
+	defer b.Unsubscribe(did, ch)
+	var mu sync.Mutex
+	var published []bus.Message
+	go func() {
+		for msg := range ch {
+			mu.Lock()
+			published = append(published, msg)
+			mu.Unlock()
+		}
+	}()
+	countPublished := func() int {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(published)
+	}
+
+	collection := syntax.NSID("place.stream.chat.message")
+	rkey := syntax.RecordKey("3lmsg000000000")
+	index := func() error {
+		bs := recCBOR
+		return atsync.handleCreateUpdate(ctx, did, rkey, &bs, rcid.String(), collection, false, false)
+	}
+
+	require.NoError(t, index())
+	require.Eventually(t, func() bool { return countPublished() == 1 }, 5*time.Second, 10*time.Millisecond,
+		"the first delivery should reach the chat bus")
+
+	// The redelivery. Same path, same CID, byte-identical record.
+	require.NoError(t, index())
+	// Give the (asynchronous) publish a chance to be wrong.
+	time.Sleep(250 * time.Millisecond)
+	require.Equal(t, 1, countPublished(), "a redelivered chat message must not be published again")
+
+	messages, err := mod.MostRecentChatMessages(did)
+	require.NoError(t, err)
+	require.Len(t, messages, 1, "a redelivered chat message must not be indexed again")
+	require.Equal(t, "hello twice", messages[0].Record.Val.(*placestream.ChatMessage).Text)
+}
+
+func TestRepoStatusFromError(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		err  error
+		want string
+	}{
+		{"nil", nil, model.RepoStatusOK},
+		{
+			"deactivated",
+			fmt.Errorf("failed to fetch verified head: %w", &xrpc.Error{
+				StatusCode: http.StatusBadRequest,
+				Wrapped:    &xrpc.XRPCError{ErrStr: "RepoDeactivated", Message: "Repo has been deactivated"},
+			}),
+			model.RepoStatusDeactivated,
+		},
+		{
+			"not found",
+			fmt.Errorf("getLatestCommit: %w", &xrpc.XRPCError{ErrStr: "RepoNotFound", Message: "Could not find repo"}),
+			model.RepoStatusNotFound,
+		},
+		{
+			"takendown",
+			fmt.Errorf("getRepo: %w", &xrpc.Error{
+				StatusCode: http.StatusBadRequest,
+				Wrapped:    &xrpc.XRPCError{ErrStr: "RepoTakendown"},
+			}),
+			model.RepoStatusTakendown,
+		},
+		{
+			// Through echo's default error handler the name arrives at the
+			// front of Message with no ErrStr at all.
+			"suspended, name only in the message",
+			fmt.Errorf("getLatestCommit: %w", &xrpc.Error{
+				StatusCode: http.StatusBadRequest,
+				Wrapped:    &xrpc.XRPCError{Message: "RepoSuspended: account suspended"},
+			}),
+			model.RepoStatusSuspended,
+		},
+		// Everything below says nothing about the account and must stay
+		// retryable: parking a repo because its host had a bad minute would
+		// take it out of the index until it happened to commit again.
+		{"block not found", fmt.Errorf("getBlocks: %w", &xrpc.XRPCError{ErrStr: "BlockNotFound"}), model.RepoStatusOK},
+		{"throttled", fmt.Errorf("getBlocks: %w", &xrpc.Error{StatusCode: http.StatusTooManyRequests}), model.RepoStatusOK},
+		{"server error", fmt.Errorf("getBlocks: %w", &xrpc.Error{StatusCode: http.StatusInternalServerError}), model.RepoStatusOK},
+		{"dns", errors.New("dial tcp: no such host"), model.RepoStatusOK},
+		{"ssrf", errors.New("request to private address blocked"), model.RepoStatusOK},
+		{"canceled", context.Canceled, model.RepoStatusOK},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			require.Equal(t, tc.want, repoStatusFromError(tc.err))
+		})
+	}
+}
+
+// TestParkTerminalRepo checks the write half: a terminal failure marks the row
+// and hands back a typed error, a transient one leaves everything alone.
+func TestParkTerminalRepo(t *testing.T) {
+	ctx := context.Background()
+	_, mod, _ := offlineSynchronizer(t)
+
+	did := "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa"
+	require.NoError(t, mod.UpdateRepo(&model.Repo{DID: did, Version: "3lrev00000000", RootCID: "bafyroot"}))
+
+	transient := fmt.Errorf("walking: %w", &xrpc.Error{StatusCode: http.StatusTooManyRequests})
+	require.Nil(t, parkTerminalRepo(ctx, mod, did, transient))
+	stored, err := mod.GetRepo(did)
+	require.NoError(t, err)
+	require.Equal(t, model.RepoStatusOK, stored.Status)
+
+	gone := fmt.Errorf("head: %w", &xrpc.XRPCError{ErrStr: "RepoDeactivated"})
+	parked := parkTerminalRepo(ctx, mod, did, gone)
+	require.Error(t, parked)
+	var terminal *TerminalRepoError
+	require.ErrorAs(t, parked, &terminal)
+	require.Equal(t, model.RepoStatusDeactivated, terminal.Status)
+	require.ErrorIs(t, parked, gone, "the underlying failure stays wrapped")
+
+	stored, err = mod.GetRepo(did)
+	require.NoError(t, err)
+	require.Equal(t, model.RepoStatusDeactivated, stored.Status)
+	require.Equal(t, "3lrev00000000", stored.Version, "whatever we indexed before stays indexed")
+	require.Equal(t, "bafyroot", stored.RootCID)
+}
+
+// TestSyncBlueskyRepoCachedSkipsTerminal: a parked row is served straight back.
+// The PDS and PLC here are ports nothing listens on, so any network attempt
+// would surface as an error rather than a returned row.
+func TestSyncBlueskyRepoCachedSkipsTerminal(t *testing.T) {
+	ctx := context.Background()
+	atsync, mod, _ := offlineSynchronizer(t)
+
+	did := "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa"
+	require.NoError(t, mod.UpdateRepo(&model.Repo{
+		DID:    did,
+		PDS:    "http://127.0.0.1:1",
+		Status: model.RepoStatusDeactivated,
+		// Deliberately no Version: without the status this row is a wedged
+		// backfill and would be re-synced immediately.
+	}))
+
+	got, err := atsync.SyncBlueskyRepoCached(ctx, did)
+	require.NoError(t, err)
+	require.Equal(t, did, got.DID)
+	require.Equal(t, model.RepoStatusDeactivated, got.Status)
+
+	// Same row without the status: now it must try, and fail.
+	require.NoError(t, mod.SetRepoStatus(ctx, did, model.RepoStatusOK))
+	_, err = atsync.SyncBlueskyRepoCached(ctx, did)
+	require.Error(t, err, "an un-parked placeholder row must actually attempt a sync")
+}
+
+// TestMigrateSkipsTerminalRepos: the boot sweep must not spend a request (or a
+// log line) on accounts that are gone.
+func TestMigrateSkipsTerminalRepos(t *testing.T) {
+	ctx := context.Background()
+	atsync, mod, _ := offlineSynchronizer(t)
+
+	did := "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa"
+	require.NoError(t, mod.UpdateRepo(&model.Repo{DID: did, PDS: "http://127.0.0.1:1"}))
+	require.NoError(t, atsync.StatefulDB.AddRepo(did))
+
+	// Control: unparked, this DID resolves nowhere, so the sweep fails on it.
+	require.Error(t, atsync.Migrate(ctx), "the only repo in the sweep should have failed")
+
+	require.NoError(t, mod.SetRepoStatus(ctx, did, model.RepoStatusDeactivated))
+	require.NoError(t, atsync.Migrate(ctx), "a terminal repo should never be dialed")
+}
+
+// TestReviveRepo: a commit proves the account is back.
+func TestReviveRepo(t *testing.T) {
+	ctx := context.Background()
+	atsync, mod, _ := offlineSynchronizer(t)
+
+	did := "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa"
+	require.NoError(t, mod.UpdateRepo(&model.Repo{DID: did, Status: model.RepoStatusTakendown, Version: "3lrev00000000"}))
+
+	r, err := mod.GetRepo(did)
+	require.NoError(t, err)
+	atsync.reviveRepo(ctx, r)
+	require.Equal(t, model.RepoStatusOK, r.Status, "the caller's copy is updated too")
+
+	stored, err := mod.GetRepo(did)
+	require.NoError(t, err)
+	require.Equal(t, model.RepoStatusOK, stored.Status)
+	require.Equal(t, "3lrev00000000", stored.Version)
+
+	// A repo that was never parked, and a repo we have never heard of, are both
+	// no-ops rather than writes.
+	atsync.reviveRepo(ctx, stored)
+	atsync.reviveRepo(ctx, nil)
+}
diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go
index 03aa1370..ce42b308 100644
--- a/pkg/atproto/sync.go
+++ b/pkg/atproto/sync.go
@@ -24,6 +24,12 @@ import (
 	glex "github.com/streamplace/glex/runtime"
 )
 
+// handleCreateUpdate indexes one record. It is called at least once per record
+// -- firehose cursor replay, a backfill walk restarting against a new head, and
+// the same commit arriving from several relays all deliver records we already
+// have -- so every write it makes has to be idempotent, and every side effect
+// (bus fanout, notification tasks) has to be skipped when nothing changed. The
+// model layer signals that with [model.ErrAlreadyIndexed]; see pkg/model/indexed.go.
 func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userDID string, rkey syntax.RecordKey, recCBOR *[]byte, cid string, collection syntax.NSID, isUpdate bool, isFirstSync bool) error {
 	ctx = log.WithLogValues(ctx, "func", "handleCreateUpdate", "userDID", userDID, "rkey", rkey.String(), "cid", cid, "collection", collection.String())
 	now := time.Now()
@@ -73,6 +79,9 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 			CID:        cid,
 		}
 		err := atsync.Model.CreateBlock(ctx, block)
+		if errors.Is(err, model.ErrAlreadyIndexed) {
+			return nil
+		}
 		if err != nil {
 			return fmt.Errorf("failed to create block: %w", err)
 		}
@@ -146,6 +155,12 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 		}
 
 		err = atsync.Model.CreateChatMessage(ctx, mcm)
+		if errors.Is(err, model.ErrAlreadyIndexed) {
+			// Already in the index: this is a cursor replay or a re-walk, not a
+			// new message. Publishing it again would show it in chat twice.
+			log.Debug(ctx, "skipping redelivered chat message", "uri", aturi.String())
+			return nil
+		}
 		if err != nil {
 			log.Error(ctx, "failed to create chat message", "err", err)
 			return nil
@@ -209,6 +224,9 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 			Repo:          repo,
 		}
 		err = atsync.Model.CreateGate(ctx, gate)
+		if errors.Is(err, model.ErrAlreadyIndexed) {
+			return nil
+		}
 		if err != nil {
 			return fmt.Errorf("failed to create gate: %w", err)
 		}
@@ -268,6 +286,9 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 			ExpiresAt:     expiresAt,
 		}
 		err = atsync.Model.CreatePinnedRecord(ctx, pin)
+		if errors.Is(err, model.ErrAlreadyIndexed) {
+			return nil
+		}
 		if err != nil {
 			return fmt.Errorf("failed to create pinned record: %w", err)
 		}
@@ -361,7 +382,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 				return fmt.Errorf("livestream url is not a string")
 			}
 			log.Debug(ctx, "livestream url", "url", url)
-			if err := atsync.Model.CreateFeedPost(ctx, &model.FeedPost{
+			err = atsync.Model.CreateFeedPost(ctx, &model.FeedPost{
 				CID:       cid,
 				CreatedAt: createdAt,
 				FeedPost:  recCBOR,
@@ -370,7 +391,8 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 				Type:      "livestream",
 				URI:       aturi.String(),
 				IndexedAt: &now,
-			}); err != nil {
+			})
+			if err != nil && !errors.Is(err, model.ErrAlreadyIndexed) {
 				return fmt.Errorf("failed to create bluesky post: %w", err)
 			}
 		} else {
@@ -417,6 +439,10 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 				IndexedAt:        &now,
 			}
 			err = atsync.Model.CreateFeedPost(ctx, fp)
+			if errors.Is(err, model.ErrAlreadyIndexed) {
+				// A reply we already have: the bus already saw it.
+				return nil
+			}
 			if err != nil {
 				log.Error(ctx, "failed to create feed post", "err", err)
 			}
@@ -449,6 +475,11 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 			ls.PostURI = rec.Post.Uri
 		}
 		err = atsync.Model.CreateLivestream(ctx, ls)
+		if errors.Is(err, model.ErrAlreadyIndexed) {
+			// Re-announcing an unchanged livestream would light the red circle
+			// up again and re-queue its finalize task.
+			return nil
+		}
 		if err != nil {
 			return fmt.Errorf("failed to create livestream: %w", err)
 		}
@@ -512,6 +543,10 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 			TargetDID:       rec.Streamer,
 		}
 		err = atsync.Model.CreateTeleport(ctx, tp)
+		if errors.Is(err, model.ErrAlreadyIndexed) {
+			// Otherwise every redelivery schedules another arrival notification.
+			return nil
+		}
 		if err != nil {
 			return fmt.Errorf("failed to create teleport: %w", err)
 		}
@@ -634,6 +669,9 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 		log.Debug(ctx, "creating moderation delegation", "streamerDID", userDID, "moderatorDID", rec.Moderator)
 
 		err = atsync.Model.CreateModerationDelegation(ctx, *rec, aturi)
+		if errors.Is(err, model.ErrAlreadyIndexed) {
+			return nil
+		}
 		if err != nil {
 			return fmt.Errorf("failed to create moderation delegation: %w", err)
 		}
@@ -865,6 +903,10 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 		}
 
 		err = atsync.Model.CreateVodComment(ctx, vc)
+		if errors.Is(err, model.ErrAlreadyIndexed) {
+			log.Debug(ctx, "skipping redelivered VOD comment", "uri", aturi.String())
+			return nil
+		}
 		if err != nil {
 			log.Error(ctx, "failed to create VOD comment", "err", err)
 			return nil
@@ -947,7 +989,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
 			Repo:          repo,
 		}
 		err = atsync.Model.CreateVodGate(ctx, gate)
-		if err != nil {
+		if err != nil && !errors.Is(err, model.ErrAlreadyIndexed) {
 			return fmt.Errorf("failed to create VOD gate: %w", err)
 		}
 
diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go
index c17f6e7d..43d5b2fa 100644
--- a/pkg/cmd/streamplace.go
+++ b/pkg/cmd/streamplace.go
@@ -267,6 +267,9 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu
 		Noter:      noter,
 		Bus:        b,
 	}
+	// Sync every repo we know about, once per boot. Nothing below depends on it
+	// having finished: it is a repair sweep for repos left half-indexed by a
+	// previous run, and the firehose keeps them current afterwards.
 	err = atsync.Migrate(ctx)
 	if err != nil {
 		return fmt.Errorf("failed to migrate: %w", err)
@@ -335,11 +338,6 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu
 	})
 	state.OATProxy = op
 
-	err = atsync.Migrate(ctx)
-	if err != nil {
-		return fmt.Errorf("failed to migrate: %w", err)
-	}
-
 	var replicator replication.Replicator = nil
 	if slices.Contains(cli.Replicators, config.ReplicatorIroh) {
 		exists, err := cli.DataFileExists([]string{"iroh-kv-secret"})
diff --git a/pkg/model/block.go b/pkg/model/block.go
index e5d57608..353950e6 100644
--- a/pkg/model/block.go
+++ b/pkg/model/block.go
@@ -50,7 +50,7 @@ func (b *Block) ToStreamplaceBlock() (placestream.Defs_BlockView, error) {
 }
 
 func (m *DBModel) CreateBlock(ctx context.Context, block *Block) error {
-	return m.DB.Create(block).Error
+	return createOrVerify(ctx, m, block, map[string]any{"rkey": block.RKey})
 }
 
 func (m *DBModel) DeleteBlock(ctx context.Context, rkey string) error {
diff --git a/pkg/model/chat_message.go b/pkg/model/chat_message.go
index 8e4e03d8..a93f9537 100644
--- a/pkg/model/chat_message.go
+++ b/pkg/model/chat_message.go
@@ -92,8 +92,11 @@ func (m *ChatMessage) ToStreamplaceMessageView() (*placestream.ChatDefs_MessageV
 	return &message, nil
 }
 
+// CreateChatMessage indexes one chat message. The table is keyed by record CID,
+// so every conflict here is a redelivery of a message we already have; an edited
+// message is a different CID and lands as its own row, as it always has.
 func (m *DBModel) CreateChatMessage(ctx context.Context, message *ChatMessage) error {
-	return m.DB.Create(message).Error
+	return createOrVerify(ctx, m, message, map[string]any{"cid": message.CID})
 }
 
 func (m *DBModel) DeleteChatMessage(ctx context.Context, uri string, deletedAt *time.Time) error {
diff --git a/pkg/model/chat_profile.go b/pkg/model/chat_profile.go
index 2cc38d2a..60d7fea0 100644
--- a/pkg/model/chat_profile.go
+++ b/pkg/model/chat_profile.go
@@ -27,6 +27,10 @@ func (m *ChatProfile) ToStreamplaceChatProfile() (placestream.ChatProfile, error
 	return scp, nil
 }
 
+// CreateChatProfile stores a user's chat profile. There is one row per repo and
+// no CID column to compare against, so Save (upsert, last write wins) is already
+// redelivery-safe: re-storing the same record rewrites the same bytes. Nothing
+// downstream fans out on it either, so it does not report ErrAlreadyIndexed.
 func (m *DBModel) CreateChatProfile(ctx context.Context, profile *ChatProfile) error {
 	err := m.DB.Save(profile).Error
 	if err != nil {
diff --git a/pkg/model/default_metadata.go b/pkg/model/default_metadata.go
index 6ecff654..d4978b64 100644
--- a/pkg/model/default_metadata.go
+++ b/pkg/model/default_metadata.go
@@ -24,6 +24,8 @@ func (m *MetadataConfiguration) ToStreamplaceMetadataConfiguration() (placestrea
 	return sdm, nil
 }
 
+// CreateMetadataConfiguration stores a repo's metadata configuration. One row
+// per repo, no CID column: Save is already redelivery-safe. See CreateChatProfile.
 func (m *DBModel) CreateMetadataConfiguration(ctx context.Context, metadata *MetadataConfiguration) error {
 	err := m.DB.Save(metadata).Error
 	if err != nil {
diff --git a/pkg/model/feed_post.go b/pkg/model/feed_post.go
index 0a8ac890..a789ba46 100644
--- a/pkg/model/feed_post.go
+++ b/pkg/model/feed_post.go
@@ -48,7 +48,7 @@ func (fp *FeedPost) ToBskyPostView() (appbsky.FeedDefs_PostView, error) {
 }
 
 func (m *DBModel) CreateFeedPost(ctx context.Context, post *FeedPost) error {
-	return m.DB.Create(post).Error
+	return createOrVerify(ctx, m, post, map[string]any{"uri": post.URI})
 }
 
 func (m *DBModel) ListFeedPosts() ([]FeedPost, error) {
diff --git a/pkg/model/follow.go b/pkg/model/follow.go
index b8ca0d6d..5d709ad2 100644
--- a/pkg/model/follow.go
+++ b/pkg/model/follow.go
@@ -16,6 +16,9 @@ type Follow struct {
 	CreatedAt  time.Time
 }
 
+// CreateFollow records a follow edge. Keyed by (user, subject) with no CID
+// column, so Save (upsert) is already redelivery-safe: the same follow arriving
+// twice rewrites the same row.
 func (m *DBModel) CreateFollow(ctx context.Context, userDID, rkey string, follow appbsky.GraphFollow) error {
 	at, err := aqtime.FromString(follow.CreatedAt)
 	if err != nil {
diff --git a/pkg/model/gate.go b/pkg/model/gate.go
index a86bd8fe..7a956872 100644
--- a/pkg/model/gate.go
+++ b/pkg/model/gate.go
@@ -26,7 +26,7 @@ func (g *Gate) ToStreamplaceGate() (placestream.ChatGate, error) {
 }
 
 func (m *DBModel) CreateGate(ctx context.Context, gate *Gate) error {
-	return m.DB.Create(gate).Error
+	return createOrVerify(ctx, m, gate, map[string]any{"rkey": gate.RKey})
 }
 
 func (m *DBModel) GetGate(ctx context.Context, rkey string) (*Gate, error) {
diff --git a/pkg/model/indexed.go b/pkg/model/indexed.go
new file mode 100644
index 00000000..0bc60f0b
--- /dev/null
+++ b/pkg/model/indexed.go
@@ -0,0 +1,121 @@
+package model
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+	"stream.place/streamplace/pkg/log"
+)
+
+// ErrAlreadyIndexed reports that a record was handed to the indexer again with
+// exactly the content the index already holds.
+//
+// Records reach us at least once, never exactly once: the firehose replays from
+// a cursor after a restart, a backfill walk that loses a race restarts against a
+// new head and re-emits everything in range, and the same commit can arrive from
+// several relays. All of that is by design, so a redelivery is not an error --
+// but it is not a fresh record either, and callers that fan a new record out to
+// the bus or to a notification queue must not do it twice. Hence a distinct
+// error rather than a plain nil: it says "nothing changed, stop here" loudly
+// enough that a caller cannot forget to check, while `err != nil` handling that
+// checks for it first stays quiet in the logs.
+var ErrAlreadyIndexed = errors.New("record already indexed")
+
+// indexedRow is one row holding one atproto record, which knows which version of
+// that record (its CID) it holds and what it is called.
+//
+// Rows keyed by CID (ChatMessage, VodComment) satisfy this trivially: for them a
+// primary-key conflict already implies the CIDs are equal, so every conflict is
+// a redelivery and there is no update arm to take.
+type indexedRow interface {
+	recordCID() string
+	recordURI() string
+}
+
+func (b *Block) recordCID() string { return b.CID }
+func (b *Block) recordURI() string {
+	return fmt.Sprintf("at://%s/app.bsky.graph.block/%s", b.RepoDID, b.RKey)
+}
+
+func (m *ChatMessage) recordCID() string { return m.CID }
+func (m *ChatMessage) recordURI() string { return m.URI }
+
+func (fp *FeedPost) recordCID() string { return fp.CID }
+func (fp *FeedPost) recordURI() string { return fp.URI }
+
+func (g *Gate) recordCID() string { return g.CID }
+func (g *Gate) recordURI() string {
+	return fmt.Sprintf("at://%s/place.stream.chat.gate/%s", g.RepoDID, g.RKey)
+}
+
+func (ls *Livestream) recordCID() string { return ls.CID }
+func (ls *Livestream) recordURI() string { return ls.URI }
+
+func (md *ModerationDelegation) recordCID() string { return md.CID }
+func (md *ModerationDelegation) recordURI() string {
+	return fmt.Sprintf("at://%s/place.stream.moderation.permission/%s", md.RepoDID, md.RKey)
+}
+
+func (p *PinnedRecord) recordCID() string { return p.CID }
+func (p *PinnedRecord) recordURI() string { return p.Uri }
+
+func (tp *Teleport) recordCID() string { return tp.CID }
+func (tp *Teleport) recordURI() string { return tp.URI }
+
+func (c *VodComment) recordCID() string { return c.CID }
+func (c *VodComment) recordURI() string { return c.URI }
+
+func (g *VodGate) recordCID() string { return g.CID }
+func (g *VodGate) recordURI() string {
+	return fmt.Sprintf("at://%s/place.stream.vod.gate/%s", g.RepoDID, g.RKey)
+}
+
+// createOrVerify writes one indexed record, idempotently.
+//
+// The insert is an ON CONFLICT DO NOTHING, so a redelivery costs one statement
+// and -- unlike the failing INSERT it replaces -- never reaches GORM's SQL error
+// logger. Only an actual conflict pays for the SELECT that decides which of the
+// two things just happened:
+//
+//   - the stored CID matches: the same record arrived twice. Nothing to write,
+//     ErrAlreadyIndexed to the caller so it skips its side effects.
+//   - the stored CID differs: an update to the record slipped past us (or a
+//     walk re-emitted a path whose record has since changed). Overwrite the row
+//     with what we were given, and let the caller treat it as news.
+//
+// key must select the conflicting row -- in practice the row's primary key.
+func createOrVerify[T any, PT interface {
+	*T
+	indexedRow
+}](ctx context.Context, m *DBModel, row PT, key map[string]any) error {
+	res := m.DB.WithContext(ctx).Clauses(clause.OnConflict{DoNothing: true}).Create(row)
+	if res.Error != nil {
+		return res.Error
+	}
+	if res.RowsAffected > 0 {
+		return nil
+	}
+
+	var existing T
+	err := m.DB.WithContext(ctx).Where(key).Take(PT(&existing)).Error
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			// The insert was refused but nothing is there to have refused it:
+			// the conflict was on some other constraint, or the row was deleted
+			// underneath us. Either way this is not the benign case.
+			return fmt.Errorf("insert of %s conflicted but no existing row matched %v", row.recordURI(), key)
+		}
+		return fmt.Errorf("failed to read conflicting row for %s: %w", row.recordURI(), err)
+	}
+
+	if PT(&existing).recordCID() == row.recordCID() {
+		return ErrAlreadyIndexed
+	}
+
+	log.Log(ctx, "record changed on redelivery, updated",
+		"uri", row.recordURI(), "oldCid", PT(&existing).recordCID(), "newCid", row.recordCID())
+	return m.DB.WithContext(ctx).Save(row).Error
+}
diff --git a/pkg/model/indexed_test.go b/pkg/model/indexed_test.go
new file mode 100644
index 00000000..4ea1bd47
--- /dev/null
+++ b/pkg/model/indexed_test.go
@@ -0,0 +1,210 @@
+package model
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"stream.place/streamplace/pkg/appbsky"
+)
+
+// indexedTestDB returns an empty in-memory index. The concrete type is handed
+// back so assertions can count rows directly, without going through the query
+// helpers (several of which filter deleted rows or need joined repos).
+func indexedTestDB(t *testing.T) *DBModel {
+	t.Helper()
+	mod, err := MakeDB(":memory:")
+	require.NoError(t, err)
+	db, ok := mod.(*DBModel)
+	require.True(t, ok)
+	return db
+}
+
+func countRows(t *testing.T, db *DBModel, model any) int64 {
+	t.Helper()
+	var n int64
+	require.NoError(t, db.DB.Model(model).Count(&n).Error)
+	return n
+}
+
+// TestCreateOrVerifyBlock covers both arms of the conflict path on a row keyed
+// by rkey with a separate CID column: the same record twice is a silent no-op,
+// and a new CID at the same key overwrites.
+func TestCreateOrVerifyBlock(t *testing.T) {
+	ctx := context.Background()
+	mod := indexedTestDB(t)
+
+	block := &Block{
+		RKey:       "3lblock1",
+		CID:        "bafyfirst",
+		RepoDID:    "did:plc:blocker",
+		SubjectDID: "did:plc:blocked",
+		Record:     []byte("one"),
+		CreatedAt:  time.Now().UTC(),
+	}
+	require.NoError(t, mod.CreateBlock(ctx, block))
+	require.Equal(t, int64(1), countRows(t, mod, &Block{}))
+
+	// Redelivery: same key, same CID.
+	again := *block
+	require.ErrorIs(t, mod.CreateBlock(ctx, &again), ErrAlreadyIndexed)
+	require.Equal(t, int64(1), countRows(t, mod, &Block{}))
+	stored, err := mod.GetBlock(ctx, block.RKey)
+	require.NoError(t, err)
+	require.Equal(t, "bafyfirst", stored.CID)
+	require.Equal(t, []byte("one"), stored.Record)
+
+	// An update we missed: same key, different CID.
+	updated := *block
+	updated.CID = "bafysecond"
+	updated.SubjectDID = "did:plc:blocked-someone-else"
+	updated.Record = []byte("two")
+	require.NoError(t, mod.CreateBlock(ctx, &updated))
+	require.Equal(t, int64(1), countRows(t, mod, &Block{}))
+	stored, err = mod.GetBlock(ctx, block.RKey)
+	require.NoError(t, err)
+	require.Equal(t, "bafysecond", stored.CID)
+	require.Equal(t, "did:plc:blocked-someone-else", stored.SubjectDID)
+	require.Equal(t, []byte("two"), stored.Record)
+}
+
+// TestCreateOrVerifyFeedPost is the same shape on a URI-keyed table.
+func TestCreateOrVerifyFeedPost(t *testing.T) {
+	ctx := context.Background()
+	mod := indexedTestDB(t)
+
+	now := time.Now().UTC()
+	body := []byte("first")
+	post := &FeedPost{
+		URI:       "at://did:plc:poster/app.bsky.feed.post/3lpost1",
+		CID:       "bafyfirst",
+		CreatedAt: now,
+		FeedPost:  &body,
+		RepoDID:   "did:plc:poster",
+		Type:      "reply",
+		IndexedAt: &now,
+	}
+	require.NoError(t, mod.CreateFeedPost(ctx, post))
+
+	again := *post
+	require.ErrorIs(t, mod.CreateFeedPost(ctx, &again), ErrAlreadyIndexed)
+	require.Equal(t, int64(1), countRows(t, mod, &FeedPost{}))
+	stored, err := mod.GetFeedPost(post.URI)
+	require.NoError(t, err)
+	require.Equal(t, "bafyfirst", stored.CID)
+	require.Equal(t, "first", string(*stored.FeedPost))
+
+	newBody := []byte("edited")
+	updated := *post
+	updated.CID = "bafysecond"
+	updated.FeedPost = &newBody
+	require.NoError(t, mod.CreateFeedPost(ctx, &updated))
+	require.Equal(t, int64(1), countRows(t, mod, &FeedPost{}))
+	stored, err = mod.GetFeedPost(post.URI)
+	require.NoError(t, err)
+	require.Equal(t, "bafysecond", stored.CID)
+	require.Equal(t, "edited", string(*stored.FeedPost))
+}
+
+// TestCreateOrVerifyChatMessage covers the CID-keyed tables. There a conflict
+// can only ever be a redelivery, and an edited message -- a different CID -- is
+// a row of its own, which is how the table has always worked.
+func TestCreateOrVerifyChatMessage(t *testing.T) {
+	ctx := context.Background()
+	mod := indexedTestDB(t)
+
+	now := time.Now().UTC()
+	body := []byte("hello")
+	msg := &ChatMessage{
+		CID:             "bafymsg1",
+		URI:             "at://did:plc:chatter/place.stream.chat.message/3lmsg1",
+		CreatedAt:       now,
+		ChatMessage:     &body,
+		RepoDID:         "did:plc:chatter",
+		StreamerRepoDID: "did:plc:streamer",
+		IndexedAt:       &now,
+	}
+	require.NoError(t, mod.CreateChatMessage(ctx, msg))
+
+	again := *msg
+	require.ErrorIs(t, mod.CreateChatMessage(ctx, &again), ErrAlreadyIndexed)
+	require.Equal(t, int64(1), countRows(t, mod, &ChatMessage{}))
+
+	edited := *msg
+	edited.CID = "bafymsg2"
+	require.NoError(t, mod.CreateChatMessage(ctx, &edited))
+	require.Equal(t, int64(2), countRows(t, mod, &ChatMessage{}),
+		"a message keyed by CID gets a new row when its content changes")
+}
+
+// TestCreateFollowIsIdempotent guards the Save-based writers, which never had a
+// duplicate problem to fix but must not grow one.
+func TestCreateFollowIsIdempotent(t *testing.T) {
+	ctx := context.Background()
+	mod := indexedTestDB(t)
+
+	follow := appbsky.GraphFollow{
+		LexiconTypeID: "app.bsky.graph.follow",
+		Subject:       "did:plc:followed",
+		CreatedAt:     time.Now().UTC().Format(time.RFC3339),
+	}
+	require.NoError(t, mod.CreateFollow(ctx, "did:plc:follower", "3lfollow1", follow))
+	require.NoError(t, mod.CreateFollow(ctx, "did:plc:follower", "3lfollow1", follow))
+	require.Equal(t, int64(1), countRows(t, mod, &Follow{}))
+
+	got, err := mod.GetUserFollowingUser(ctx, "did:plc:follower", "did:plc:followed")
+	require.NoError(t, err)
+	require.NotNil(t, got)
+	require.Equal(t, "3lfollow1", got.RKey)
+}
+
+// TestCreateChatProfileIsIdempotent: one row per repo, no CID to compare, so
+// re-storing it is a plain overwrite rather than ErrAlreadyIndexed.
+func TestCreateChatProfileIsIdempotent(t *testing.T) {
+	ctx := context.Background()
+	mod := indexedTestDB(t)
+
+	rec := []byte("profile")
+	profile := &ChatProfile{RepoDID: "did:plc:chatter", Record: &rec}
+	require.NoError(t, mod.CreateChatProfile(ctx, profile))
+	require.NoError(t, mod.CreateChatProfile(ctx, profile))
+	require.Equal(t, int64(1), countRows(t, mod, &ChatProfile{}))
+}
+
+// TestRepoStatus covers the account lifecycle column: setting a status leaves
+// the sync state alone, and the terminal list is what the boot sweep filters on.
+func TestRepoStatus(t *testing.T) {
+	ctx := context.Background()
+	mod := indexedTestDB(t)
+
+	require.NoError(t, mod.UpdateRepo(&Repo{
+		DID:     "did:plc:gone",
+		Handle:  "gone.test",
+		PDS:     "https://pds.test",
+		Version: "3lrev0000",
+		RootCID: "bafyroot",
+	}))
+	require.NoError(t, mod.UpdateRepo(&Repo{DID: "did:plc:fine", Version: "3lrev0001"}))
+
+	dids, err := mod.TerminalRepoDIDs(ctx)
+	require.NoError(t, err)
+	require.Empty(t, dids)
+
+	require.NoError(t, mod.SetRepoStatus(ctx, "did:plc:gone", RepoStatusDeactivated))
+	stored, err := mod.GetRepo("did:plc:gone")
+	require.NoError(t, err)
+	require.Equal(t, RepoStatusDeactivated, stored.Status)
+	require.True(t, stored.TerminalStatus())
+	require.Equal(t, "3lrev0000", stored.Version, "status must not disturb the sync state")
+	require.Equal(t, "bafyroot", stored.RootCID)
+
+	dids, err = mod.TerminalRepoDIDs(ctx)
+	require.NoError(t, err)
+	require.Equal(t, []string{"did:plc:gone"}, dids)
+
+	require.NoError(t, mod.SetRepoStatus(ctx, "did:plc:gone", RepoStatusOK))
+	dids, err = mod.TerminalRepoDIDs(ctx)
+	require.NoError(t, err)
+	require.Empty(t, dids)
+}
diff --git a/pkg/model/livestream.go b/pkg/model/livestream.go
index 76bf1109..dbd570f2 100644
--- a/pkg/model/livestream.go
+++ b/pkg/model/livestream.go
@@ -8,7 +8,6 @@ import (
 
 	glex "github.com/streamplace/glex/runtime"
 	"gorm.io/gorm"
-	"gorm.io/gorm/clause"
 	"stream.place/streamplace/pkg/appbsky"
 	"stream.place/streamplace/pkg/moderation"
 	"stream.place/streamplace/pkg/placestream"
@@ -49,12 +48,12 @@ func (ls *Livestream) ToLivestreamView() (*placestream.Livestream_LivestreamView
 	return &postView, nil
 }
 
+// CreateLivestream upserts a livestream record. It used to overwrite the row on
+// every conflict, which meant a redelivered record -- the same bytes, the same
+// CID -- rewrote the row and re-announced the stream on the bus. Now an
+// unchanged record is a no-op and only a genuinely new version is stored.
 func (m *DBModel) CreateLivestream(ctx context.Context, ls *Livestream) error {
-	// upsert livestream record, actually
-	return m.DB.Clauses(clause.OnConflict{
-		Columns:   []clause.Column{{Name: "uri"}},
-		DoUpdates: clause.AssignmentColumns([]string{"cid", "created_at", "livestream", "repo_did", "post_cid", "post_uri"}),
-	}).Create(ls).Error
+	return createOrVerify(ctx, m, ls, map[string]any{"uri": ls.URI})
 }
 
 func (m *DBModel) GetLivestream(uri string) (*Livestream, error) {
diff --git a/pkg/model/model.go b/pkg/model/model.go
index ad23ff8e..75869bdb 100644
--- a/pkg/model/model.go
+++ b/pkg/model/model.go
@@ -38,6 +38,8 @@ type Model interface {
 	GetAllRepos() ([]Repo, error)
 	SearchReposByHandle(query string, limit int) ([]Repo, error)
 	UpdateRepo(repo *Repo) error
+	SetRepoStatus(ctx context.Context, did string, status string) error
+	TerminalRepoDIDs(ctx context.Context) ([]string, error)
 
 	UpdateSigningKey(key *SigningKey) error
 	GetSigningKey(ctx context.Context, did, repoDID string) (*SigningKey, error)
diff --git a/pkg/model/moderation_delegation.go b/pkg/model/moderation_delegation.go
index 561a05cc..651fbd6e 100644
--- a/pkg/model/moderation_delegation.go
+++ b/pkg/model/moderation_delegation.go
@@ -80,7 +80,7 @@ func (m *DBModel) CreateModerationDelegation(ctx context.Context, rec placestrea
 		IndexedAt:    now.Time().UTC(),
 	}
 
-	return m.DB.WithContext(ctx).Create(delegation).Error
+	return createOrVerify(ctx, m, delegation, map[string]any{"rkey": rkey})
 }
 
 func (m *DBModel) DeleteModerationDelegation(ctx context.Context, rkey string) error {
diff --git a/pkg/model/pinned_record.go b/pkg/model/pinned_record.go
index 121b10da..a541aa4d 100644
--- a/pkg/model/pinned_record.go
+++ b/pkg/model/pinned_record.go
@@ -56,7 +56,7 @@ func (p *PinnedRecord) ToStreamplacePinnedRecordView() (placestream.ChatDefs_Pin
 	return rec, nil
 }
 func (m *DBModel) CreatePinnedRecord(ctx context.Context, pin *PinnedRecord) error {
-	return m.DB.Create(pin).Error
+	return createOrVerify(ctx, m, pin, map[string]any{"uri": pin.Uri})
 }
 
 func (m *DBModel) GetPinnedRecord(ctx context.Context, uri string) (*PinnedRecord, error) {
diff --git a/pkg/model/repo.go b/pkg/model/repo.go
index 44735a9c..92dbc599 100644
--- a/pkg/model/repo.go
+++ b/pkg/model/repo.go
@@ -1,17 +1,38 @@
 package model
 
 import (
+	"context"
 	"errors"
 
 	"gorm.io/gorm"
 )
 
+// Account lifecycle states a repo row can be parked in. Anything other than
+// RepoStatusOK is terminal: the account is gone, hidden, or turned off, so
+// retrying its backfill on every boot only burns requests. A live commit on the
+// firehose is what proves the account is back and clears it.
+const (
+	RepoStatusOK          = ""
+	RepoStatusDeactivated = "deactivated"
+	RepoStatusNotFound    = "notfound"
+	RepoStatusTakendown   = "takendown"
+	RepoStatusSuspended   = "suspended"
+)
+
 type Repo struct {
 	DID     string `gorm:"primaryKey;column:did" json:"did"`
 	Handle  string `gorm:"index" json:"handle"`
 	PDS     string `json:"pds"`
 	Version string `json:"version"`
 	RootCID string `json:"rootCid"`
+	// Status is one of the RepoStatus* constants; empty for a normal account.
+	Status string `gorm:"column:status" json:"status,omitempty"`
+}
+
+// TerminalStatus reports whether this repo is in an account state no amount of
+// retrying will get us past.
+func (r *Repo) TerminalStatus() bool {
+	return r != nil && r.Status != RepoStatusOK
 }
 
 func (Repo) TableName() string {
@@ -78,6 +99,25 @@ func (m *DBModel) UpdateRepo(repo *Repo) error {
 	return m.DB.Save(repo).Error
 }
 
+// SetRepoStatus parks (or un-parks) a repo's account lifecycle state without
+// touching the sync state in the rest of the row.
+func (m *DBModel) SetRepoStatus(ctx context.Context, did string, status string) error {
+	return m.DB.WithContext(ctx).Model(&Repo{}).Where("did = ?", did).Update("status", status).Error
+}
+
+// TerminalRepoDIDs lists the repos parked in a terminal account state, so the
+// boot-time sync sweep can skip them in one query instead of failing on each.
+func (m *DBModel) TerminalRepoDIDs(ctx context.Context) ([]string, error) {
+	var dids []string
+	err := m.DB.WithContext(ctx).Model(&Repo{}).
+		Where("status IS NOT NULL AND status != ?", RepoStatusOK).
+		Pluck("did", &dids).Error
+	if err != nil {
+		return nil, err
+	}
+	return dids, nil
+}
+
 func (m *DBModel) SearchReposByHandle(query string, limit int) ([]Repo, error) {
 	var repos []Repo
 	// Search for repos where handle starts with the query (case-insensitive)
diff --git a/pkg/model/teleport.go b/pkg/model/teleport.go
index eed76d91..3693aa4b 100644
--- a/pkg/model/teleport.go
+++ b/pkg/model/teleport.go
@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"gorm.io/gorm"
-	"gorm.io/gorm/clause"
 )
 
 type Teleport struct {
@@ -24,11 +23,11 @@ type Teleport struct {
 	Target          *Repo     `json:"target,omitempty" gorm:"foreignKey:DID;references:TargetDID"`
 }
 
+// CreateTeleport upserts a teleport record. As with livestreams, an unchanged
+// record is now a no-op: re-indexing one used to reschedule its arrival
+// notification and re-publish it to the bus.
 func (m *DBModel) CreateTeleport(ctx context.Context, tp *Teleport) error {
-	return m.DB.Clauses(clause.OnConflict{
-		Columns:   []clause.Column{{Name: "uri"}},
-		DoUpdates: clause.AssignmentColumns([]string{"cid", "starts_at", "duration_seconds", "viewer_count", "teleport", "repo_did", "target_did"}),
-	}).Create(tp).Error
+	return createOrVerify(ctx, m, tp, map[string]any{"uri": tp.URI})
 }
 
 func (m *DBModel) GetLatestTeleportForRepo(repoDID string) (*Teleport, error) {
diff --git a/pkg/model/vod_comment.go b/pkg/model/vod_comment.go
index 8933a8f1..e1d18eee 100644
--- a/pkg/model/vod_comment.go
+++ b/pkg/model/vod_comment.go
@@ -100,8 +100,10 @@ func (c *VodComment) ToStreamplaceCommentView() (placestream.VodDefs_CommentView
 	return commentView, nil
 }
 
+// CreateVodComment indexes one VOD comment. Like chat messages the table is
+// keyed by record CID, so a conflict is always a redelivery of the same comment.
 func (m *DBModel) CreateVodComment(ctx context.Context, comment *VodComment) error {
-	return m.DB.Create(comment).Error
+	return createOrVerify(ctx, m, comment, map[string]any{"cid": comment.CID})
 }
 
 func (m *DBModel) DeleteVodComment(ctx context.Context, uri string, deletedAt *time.Time) error {
diff --git a/pkg/model/vod_gate.go b/pkg/model/vod_gate.go
index 491df90e..45e23594 100644
--- a/pkg/model/vod_gate.go
+++ b/pkg/model/vod_gate.go
@@ -26,7 +26,7 @@ func (g *VodGate) ToStreamplaceVodGate() (placestream.VodGate, error) {
 }
 
 func (m *DBModel) CreateVodGate(ctx context.Context, gate *VodGate) error {
-	return m.DB.Create(gate).Error
+	return createOrVerify(ctx, m, gate, map[string]any{"rkey": gate.RKey})
 }
 
 func (m *DBModel) GetVodGate(ctx context.Context, rkey string) (*VodGate, error) {