Merge pull request #456 from streamplace/eli/postgres

statedb: migrate to a stateful database for stateful things

.vscode/launch.json

···
+{
+  "version": "0.2.0",
+  "configurations": [
+    {
+      "name": "Run Streamplace (linux)",
+      "type": "go",
+      "request": "launch",
+      "mode": "exec",
+      "program": "${workspaceFolder}/build-linux-amd64/libstreamplace"
+    },
+    {
+      "name": "Run Streamplace (darwin)",
+      "type": "go",
+      "request": "launch",
+      "mode": "exec",
+      "program": "${workspaceFolder}/build-darwin-amd64/libstreamplace"
+    }
+  ]
+}

go.mod

···
 
 replace github.com/AxisCommunications/go-dpop => github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4
 
+replace github.com/bluesky-social/indigo => github.com/streamplace/indigo v0.0.0-20250813192504-b19ccd82854b
+
 require (
 	firebase.google.com/go/v4 v4.14.1
 	git.stream.place/streamplace/c2pa-go v0.7.0
···
 	github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e
 	github.com/slok/go-http-metrics v0.13.0
 	github.com/streamplace/atproto-oauth-golang v0.0.0-20250619231223-a9c04fb888ac
-	github.com/streamplace/oatproxy v0.0.0-20250722215839-1c878921d185
+	github.com/streamplace/oatproxy v0.0.0-20250813225644-39354b83af26
 	github.com/stretchr/testify v1.10.0
 	github.com/tdewolff/canvas v0.0.0-20250728095813-50d4cb1eee71
 	github.com/whyrusleeping/cbor-gen v0.3.1
···
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da
 	google.golang.org/api v0.228.0
 	gorm.io/datatypes v1.2.4
+	gorm.io/driver/postgres v1.5.7
 	gorm.io/driver/sqlite v1.5.7
 )
 
···
 	gopkg.in/warnings.v0 v0.1.2 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gorm.io/driver/mysql v1.5.6 // indirect
-	gorm.io/driver/postgres v1.5.7 // indirect
 	gotest.tools/v3 v3.5.2 // indirect
 	honnef.co/go/tools v0.6.1 // indirect
 	howett.net/plist v1.0.0 // indirect

go.sum

···
 github.com/bluenviron/gortsplib/v4 v4.12.3/go.mod h1:SkZPdaMNr+IvHt2PKRjUXxZN6FDutmSZn4eT0GmF0sk=
 github.com/bluenviron/mediacommon/v2 v2.4.0 h1:Ss1T7AMxTrICJ+a/N5urS/1lp1ZpsF+3iJq3B/RLDMw=
 github.com/bluenviron/mediacommon/v2 v2.4.0/go.mod h1:a6MbPmXtYda9mKibKVMZlW20GYLLrX2R7ZkUE+1pwV0=
-github.com/bluesky-social/indigo v0.0.0-20250729223159-573ae927246a h1:S12KN45uIkRglMHC8PqD/Vsz0+u3KbIaBF/6rit8/Pg=
-github.com/bluesky-social/indigo v0.0.0-20250729223159-573ae927246a/go.mod h1:0XUyOCRtL4/OiyeqMTmr6RlVHQMDgw3LS7CfibuZR5Q=
 github.com/bmatcuk/doublestar v1.1.1/go.mod h1:UD6OnuiIn0yFxxA2le/rnRU1G4RaI4UvFv1sNto9p6w=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
···
 github.com/streamplace/atproto-oauth-golang v0.0.0-20250619231223-a9c04fb888ac/go.mod h1:9LlKkqciiO5lRfbX0n4Wn5KNY9nvFb4R3by8FdW2TWc=
 github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4 h1:L1fS4HJSaAyNnkwfuZubgfeZy8rkWmA0cMtH5Z0HqNc=
 github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4/go.mod h1:bGUXY9Wd4mnd+XUrOYZr358J2f6z9QO/dLhL1SsiD+0=
+github.com/streamplace/indigo v0.0.0-20250813192504-b19ccd82854b h1:vh0Z1bDgcVXQjgmcXOBkC3B4uWi+HaxeyTvETzcA/Vg=
+github.com/streamplace/indigo v0.0.0-20250813192504-b19ccd82854b/go.mod h1:n6QE1NDPFoi7PRbMUZmc2y7FibCqiVU4ePpsvhHUBR8=
 github.com/streamplace/oatproxy v0.0.0-20250722215839-1c878921d185 h1:knTi1I/zbppb+CmbDRmPKNgqO4QPOKdZBds63mMWTms=
 github.com/streamplace/oatproxy v0.0.0-20250722215839-1c878921d185/go.mod h1:pXi24hA7xBHj8eEywX6wGqJOR9FaEYlGwQ/72rN6okw=
+github.com/streamplace/oatproxy v0.0.0-20250813225644-39354b83af26 h1:tbGie3B0tbjp47BnG0sAhjlYLZXHBbEEW8chdbcbrn4=
+github.com/streamplace/oatproxy v0.0.0-20250813225644-39354b83af26/go.mod h1:pXi24hA7xBHj8eEywX6wGqJOR9FaEYlGwQ/72rN6okw=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=

pkg/api/api.go

···
 	"stream.place/streamplace/pkg/notifications"
 	"stream.place/streamplace/pkg/spmetrics"
 	"stream.place/streamplace/pkg/spxrpc"
+	"stream.place/streamplace/pkg/statedb"
 	"stream.place/streamplace/pkg/streamplace"
 
 	metrics "github.com/slok/go-http-metrics/metrics/prometheus"
···
 type StreamplaceAPI struct {
 	CLI              *config.CLI
 	Model            model.Model
+	StatefulDB       *statedb.StatefulDB
 	Updater          *Updater
 	Signer           *eip712.EIP712Signer
 	Mimes            map[string]string
···
 	mu            sync.RWMutex
 }
 
-func MakeStreamplaceAPI(cli *config.CLI, mod model.Model, signer *eip712.EIP712Signer, noter notifications.FirebaseNotifier, mm *media.MediaManager, ms media.MediaSigner, bus *bus.Bus, atsync *atproto.ATProtoSynchronizer, d *director.Director, op *oatproxy.OATProxy) (*StreamplaceAPI, error) {
+func MakeStreamplaceAPI(cli *config.CLI, mod model.Model, statefulDB *statedb.StatefulDB, signer *eip712.EIP712Signer, noter notifications.FirebaseNotifier, mm *media.MediaManager, ms media.MediaSigner, bus *bus.Bus, atsync *atproto.ATProtoSynchronizer, d *director.Director, op *oatproxy.OATProxy) (*StreamplaceAPI, error) {
 	updater, err := PrepareUpdater(cli)
 	if err != nil {
 		return nil, err
 	}
 	a := &StreamplaceAPI{CLI: cli,
 		Model:            mod,
+		StatefulDB:       statefulDB,
 		Updater:          updater,
 		Signer:           signer,
 		FirebaseNotifier: noter,
···
 		Recorder: metrics.NewRecorder(metrics.Config{}),
 	})
 	var xrpc http.Handler
-	xrpc, err := spxrpc.NewServer(ctx, a.CLI, a.Model, a.op, mdlw, a.ATSync)
+	xrpc, err := spxrpc.NewServer(ctx, a.CLI, a.Model, a.StatefulDB, a.op, mdlw, a.ATSync)
 	if err != nil {
 		return nil, err
 	}
···
 			w.WriteHeader(400)
 			return
 		}
-		err = a.Model.CreateNotification(n.Token, n.RepoDID)
+		err = a.StatefulDB.CreateNotification(n.Token, n.RepoDID)
 		if err != nil {
 			log.Log(ctx, "error creating notification", "error", err)
 			w.WriteHeader(400)

pkg/api/api_internal.go

···
 	})
 
 	router.GET("/notifications", func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
-		notifications, err := a.Model.ListNotifications()
+		notifications, err := a.StatefulDB.ListNotifications()
 		if err != nil {
 			errors.WriteHTTPInternalServerError(w, "unable to get notifications", err)
 			return
···
 	})
 
 	router.GET("/oauth-sessions", func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
-		sessions, err := a.Model.ListOAuthSessions()
+		sessions, err := a.StatefulDB.ListOAuthSessions()
 		if err != nil {
 			errors.WriteHTTPInternalServerError(w, "unable to get oauth sessions", err)
 			return
···
 			errors.WriteHTTPBadRequest(w, "invalid request body", err)
 			return
 		}
-		notifications, err := a.Model.ListNotifications()
+		notifications, err := a.StatefulDB.ListNotifications()
 		if err != nil {
 			errors.WriteHTTPInternalServerError(w, "unable to get notifications", err)
 			return

pkg/atproto/firehose.go

···
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/model"
 	notificationpkg "stream.place/streamplace/pkg/notifications"
+	"stream.place/streamplace/pkg/statedb"
 
 	"slices"
 
···
 )
 
 type ATProtoSynchronizer struct {
-	CLI       *config.CLI
-	Model     model.Model
-	LastSeen  time.Time
-	LastEvent time.Time
-	Noter     notificationpkg.FirebaseNotifier
-	Bus       *bus.Bus
+	CLI        *config.CLI
+	Model      model.Model
+	StatefulDB *statedb.StatefulDB
+	LastSeen   time.Time
+	LastEvent  time.Time
+	Noter      notificationpkg.FirebaseNotifier
+	Bus        *bus.Bus
 }
 
 func (atsync *ATProtoSynchronizer) StartFirehose(ctx context.Context) error {

pkg/atproto/jwks.go

···
-package atproto
-
-import (
-	"context"
-	"encoding/json"
-	"os"
-
-	"github.com/lestrrat-go/jwx/v2/jwk"
-	oauth_helpers "github.com/streamplace/atproto-oauth-golang/helpers"
-	"stream.place/streamplace/pkg/log"
-)
-
-func EnsureJWK(ctx context.Context, fPath string) (jwk.Key, error) {
-	var key jwk.Key
-	_, err := os.Stat(fPath)
-	if err == nil {
-		b, err := os.ReadFile(fPath)
-		if err != nil {
-			return nil, err
-		}
-		key, err = jwk.ParseKey(b)
-		if err != nil {
-			return nil, err
-		}
-	} else if os.IsNotExist(err) {
-		key, err = oauth_helpers.GenerateKey(nil)
-		if err != nil {
-			return nil, err
-		}
-
-		b, err := json.Marshal(key)
-		if err != nil {
-			return nil, err
-		}
-
-		if err := os.WriteFile(fPath, b, 0600); err != nil {
-			return nil, err
-		}
-		log.Log(ctx, "generated JWK", "path", fPath)
-	} else {
-		return nil, err
-	}
-
-	return key, nil
-}

pkg/atproto/lexicon_repo.go

···
 	"github.com/bluesky-social/indigo/util"
 	"github.com/ipfs/go-cid"
 	cbg "github.com/whyrusleeping/cbor-gen"
-	"gorm.io/driver/sqlite"
-	"gorm.io/gorm"
 
 	"github.com/whyrusleeping/go-did"
 	"stream.place/streamplace/lexicons"
 	"stream.place/streamplace/pkg/config"
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/model"
+	"stream.place/streamplace/pkg/statedb"
 )
 
 var LexiconRepo *atrepo.Repo
···
 	Close() error
 }
 
-func MakeLexiconRepo(ctx context.Context, cli *config.CLI, mod model.Model) (Closer, error) {
+type NoopCloser struct{}
+
+func (c *NoopCloser) Close() error {
+	return nil
+}
+
+func MakeLexiconRepo(ctx context.Context, cli *config.CLI, mod model.Model, state *statedb.StatefulDB) (Closer, error) {
 	ctx = log.WithLogValues(ctx, "func", "MakeLexiconRepo")
-	fd, err := cli.DataFileCreate([]string{"carstore", "empty"}, true)
-	if err != nil {
-		return nil, err
-	}
-	sqlitePath := cli.DataFilePath([]string{"carstore", "meta.sqlite"})
+	var err error
 
-	db, err := gorm.Open(sqlite.Open(sqlitePath))
-	if err != nil {
-		return nil, err
-	}
-	err = fd.Close()
-	if err != nil {
-		return nil, err
-	}
-	CarStore, err = carstore.NewCarStore(db, []string{
-		cli.DataFilePath([]string{"carstore"}),
-	})
-	if err != nil {
-		return nil, err
-	}
+	sqliteStore := &carstore.SQLiteStore{}
 
-	sqlDB, err := db.DB()
+	err = sqliteStore.Open(":memory:")
 	if err != nil {
 		return nil, err
 	}
+	CarStore = sqliteStore
 
 	var priv *atcrypto.PrivateKeyK256
-	exists, err := cli.DataFileExists([]string{"carstore", "repo.key"})
+
+	keyBs, err := state.GetConfig("repo-key")
 	if err != nil {
 		return nil, err
 	}
-	if exists {
-		buf := bytes.Buffer{}
-		err := cli.DataFileRead([]string{"carstore", "repo.key"}, &buf)
+	if keyBs != nil {
+		// good path: we have a key in the stateful database
+		priv, err = atcrypto.ParsePrivateBytesK256(keyBs.Value)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to parse repo key from stateful database: %w", err)
 		}
-		priv, err = atcrypto.ParsePrivateBytesK256(buf.Bytes())
+	} else {
+		// migration path: maybe we have an old one on disk.
+		exists, err := cli.DataFileExists([]string{"carstore", "repo.key"})
 		if err != nil {
 			return nil, err
 		}
-	} else {
-		priv, err = atcrypto.GeneratePrivateKeyK256()
-		if err != nil {
-			return nil, err
+		if exists {
+			log.Warn(ctx, "found old repo key on disk, migrating to stateful database", "path", cli.DataFilePath([]string{"carstore", "repo.key"}))
+			buf := bytes.Buffer{}
+			err := cli.DataFileRead([]string{"carstore", "repo.key"}, &buf)
+			if err != nil {
+				return nil, err
+			}
+			priv, err = atcrypto.ParsePrivateBytesK256(buf.Bytes())
+			if err != nil {
+				return nil, fmt.Errorf("failed to read repo key from disk: %w", err)
+			}
+		} else {
+			priv, err = atcrypto.GeneratePrivateKeyK256()
+			if err != nil {
+				return nil, err
+			}
 		}
 		bs := priv.Bytes()
-		err = cli.DataFileWrite([]string{"carstore", "repo.key"}, bytes.NewReader(bs), true)
+		err = state.PutConfig("repo-key", bs)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to save repo key to stateful database: %w", err)
 		}
 	}
 
···
 		return priv.HashAndSign(sb)
 	}
 
-	ses, err := CarStore.NewDeltaSession(ctx, RepoUser, nil)
+	events, err := state.GetCommitEventsSince(cli.MyDID(), time.Time{})
 	if err != nil {
-		return nil, fmt.Errorf("failed to create delta session: %w", err)
+		return nil, fmt.Errorf("failed to get commit events: %w", err)
 	}
 
-	currentRoot, err := CarStore.GetUserRepoHead(ctx, RepoUser)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get user repo head: %w", err)
+	var ses *carstore.DeltaSession
+	var currentRoot cid.Cid
+	var currentRev string
+
+	for _, event := range events {
+		evt, err := event.ToCommitEvent()
+		if err != nil {
+			return nil, fmt.Errorf("failed to convert event to commit event: %w", err)
+		}
+		currentRoot, ses, err = CarStore.ImportSlice(ctx, RepoUser, nil, evt.Blocks)
+		if err != nil {
+			return nil, fmt.Errorf("failed to import slice: %w", err)
+		}
+		currentRev = evt.Rev
 	}
-	currentRev := ""
 
 	if currentRoot == cid.Undef {
+		log.Warn(ctx, "no existing lexicon repo, creating new one")
+		ses, err = CarStore.NewDeltaSession(ctx, RepoUser, nil)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create delta session: %w", err)
+		}
 		LexiconRepo = atrepo.NewRepo(ctx, cli.MyDID(), ses)
 	} else {
 		LexiconRepo, err = atrepo.OpenRepo(ctx, ses, currentRoot)
 		if err != nil {
 			return nil, fmt.Errorf("failed to open repo: %w", err)
-		}
-		currentRev, err = CarStore.GetUserRepoRev(ctx, RepoUser)
-		if err != nil {
-			return nil, fmt.Errorf("failed to get user repo rev: %w", err)
 		}
 	}
 
···
 			Ops:    ops,
 			TooBig: false,
 		}
-		err := mod.CreateCommitEvent(commit, signed.Data.String())
+		err := state.CreateCommitEvent(commit, signed.Data.String())
 		if err != nil {
 			return nil, fmt.Errorf("failed to create commit event: %w", err)
 		}
 	}
 
-	return sqlDB, nil
+	return &NoopCloser{}, nil
 }
 
 func OpenLexiconRepo(ctx context.Context) (*atrepo.Repo, *carstore.DeltaSession, error) {

pkg/atproto/lexicon_repo_test.go

···
 	"stream.place/streamplace/lexicons"
 	"stream.place/streamplace/pkg/config"
 	"stream.place/streamplace/pkg/model"
+	"stream.place/streamplace/pkg/statedb"
 )
 
 func TestLexiconRepo(t *testing.T) {
 	cli := config.CLI{
 		PublicHost: "example.com",
+		DBURL:      ":memory:",
 	}
 	cli.DataDir = t.TempDir()
 	mod, err := model.MakeDB(":memory:")
 	require.NoError(t, err)
+	state, err := statedb.MakeDB(&cli, nil, mod)
+	require.NoError(t, err)
 
 	// creating a new repo
-	handle, err := MakeLexiconRepo(context.Background(), &cli, mod)
+	handle, err := MakeLexiconRepo(context.Background(), &cli, mod, state)
 	require.NoError(t, err)
 	r, sess, err := OpenLexiconRepo(context.Background())
 	require.NoError(t, err)
···
 	require.NotNil(t, rec)
 	handle.Close()
 
-	evts, err := mod.GetCommitEventsSinceSeq(cli.MyDID(), 0)
+	evts, err := state.GetCommitEventsSinceSeq(cli.MyDID(), 0)
 	require.NoError(t, err)
 	require.Len(t, evts, 1)
 	require.Equal(t, evts[0].RepoDID, cli.MyDID())
 
 	// opening an existing repo
-	handle, err = MakeLexiconRepo(context.Background(), &cli, mod)
+	handle, err = MakeLexiconRepo(context.Background(), &cli, mod, state)
 	require.NoError(t, err)
 	handle.Close()
 
···
 	AllFiles = modifiedFS
 
 	// opening an existing repo with modified lexicon
-	handle, err = MakeLexiconRepo(context.Background(), &cli, mod)
+	handle, err = MakeLexiconRepo(context.Background(), &cli, mod, state)
 	require.NoError(t, err)
 	handle.Close()
 
-	evts, err = mod.GetCommitEventsSinceSeq(cli.MyDID(), 0)
+	evts, err = state.GetCommitEventsSinceSeq(cli.MyDID(), 0)
 	require.NoError(t, err)
 	require.Len(t, evts, 2)
 	require.Equal(t, evts[0].RepoDID, cli.MyDID())

pkg/atproto/sync.go

···
 	"github.com/bluesky-social/indigo/atproto/data"
 	"github.com/bluesky-social/indigo/atproto/syntax"
 	"stream.place/streamplace/pkg/aqtime"
-	"stream.place/streamplace/pkg/integrations/discord"
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/model"
-	notificationpkg "stream.place/streamplace/pkg/notifications"
+	"stream.place/streamplace/pkg/statedb"
 	"stream.place/streamplace/pkg/streamplace"
 
 	lexutil "github.com/bluesky-social/indigo/lex/util"
···
 			return fmt.Errorf("failed to sync bluesky repo: %w", err)
 		}
 
-		_, err = atsync.SyncBlueskyRepoCached(ctx, rec.Streamer, atsync.Model)
-		if err != nil {
-			log.Error(ctx, "failed to sync bluesky repo", "err", err)
-		}
+		go func() {
+			_, err = atsync.SyncBlueskyRepoCached(ctx, rec.Streamer, atsync.Model)
+			if err != nil {
+				log.Error(ctx, "failed to sync bluesky repo", "err", err)
+			}
+		}()
 
 		log.Debug(ctx, "streamplace.ChatMessage detected", "message", rec.Text, "repo", repo.Handle)
 		block, err := atsync.Model.GetUserBlock(ctx, rec.Streamer, userDID)
···
 		go atsync.Bus.Publish(rec.Streamer, scm)
 
 		if !isUpdate && !isFirstSync {
-			for _, webhook := range atsync.CLI.DiscordWebhooks {
-				if webhook.DID == rec.Streamer && webhook.Type == "chat" {
-					go func() {
-						err := discord.SendChat(ctx, webhook, repo, scm)
-						if err != nil {
-							log.Error(ctx, "failed to send livestream to discord", "err", err)
-						} else {
-							log.Log(ctx, "sent livestream to discord", "user", userDID, "webhook", webhook.URL)
-						}
-					}()
-				}
+
+			task := &statedb.ChatTask{
+				MessageView: scm,
+			}
+
+			_, err = atsync.StatefulDB.EnqueueTask(ctx, statedb.TaskChat, task, statedb.WithTaskKey(fmt.Sprintf("chat-message::%s", aturi.String())))
+			if err != nil {
+				log.Error(ctx, "failed to enqueue notification task", "err", err)
 			}
 		}
 
···
 		}
 
 	case *streamplace.Livestream:
-		var u string
-		if rec.Url != nil {
-			u = *rec.Url
-		}
 		if r == nil {
 			// we don't know about this repo
 			return nil
···
 		}
 		go atsync.Bus.Publish(userDID, lsv)
 
-		if !isUpdate && !isFirstSync {
-			log.Warn(ctx, "Livestream detected! Blasting followers!", "title", rec.Title, "url", u, "createdAt", rec.CreatedAt, "repo", userDID)
-			notifications, err := atsync.Model.GetFollowersNotificationTokens(userDID)
+		var postView *bsky.FeedDefs_PostView
+		if lsHydrated.Post != nil {
+			postView, err = lsHydrated.Post.ToBskyPostView()
 			if err != nil {
-				return err
-			}
-
-			nb := &notificationpkg.NotificationBlast{
-				Title: fmt.Sprintf("🔴 @%s is LIVE!", r.Handle),
-				Body:  rec.Title,
-				Data: map[string]string{
-					"path": fmt.Sprintf("/%s", r.Handle),
-				},
-			}
-			if atsync.Noter != nil {
-				err := atsync.Noter.Blast(ctx, notifications, nb)
-				if err != nil {
-					log.Error(ctx, "failed to blast notifications", "err", err)
-				} else {
-					log.Log(ctx, "sent notifications", "user", userDID, "count", len(notifications), "content", nb)
-				}
-			} else {
-				log.Log(ctx, "no notifier configured, skipping notifications", "user", userDID, "count", len(notifications), "content", nb)
+				return fmt.Errorf("failed to convert livestream post to bsky post view: %w", err)
 			}
+		}
 
-			var postView *bsky.FeedDefs_PostView
-			if lsHydrated.Post != nil {
-				postView, err = lsHydrated.Post.ToBskyPostView()
-				if err != nil {
-					log.Error(ctx, "failed to convert livestream post to bsky post view", "err", err)
-				}
-			} else {
-				log.Warn(ctx, "no post found for livestream", "livestream", lsHydrated)
-			}
+		task := &statedb.NotificationTask{
+			Livestream: lsv,
+			FeedPost:   postView,
+			PDSURL:     r.PDS,
+		}
 
-			var spcp *streamplace.ChatProfile
-			cp, err := atsync.Model.GetChatProfile(ctx, userDID)
+		cp, err := atsync.Model.GetChatProfile(ctx, userDID)
+		if err != nil {
+			return fmt.Errorf("failed to get chat profile: %w", err)
+		}
+		if cp != nil {
+			spcp, err := cp.ToStreamplaceChatProfile()
 			if err != nil {
-				log.Error(ctx, "failed to get chat profile", "err", err)
-			}
-			if cp != nil {
-				spcp, err = cp.ToStreamplaceChatProfile()
-				if err != nil {
-					log.Error(ctx, "failed to convert chat profile to streamplace chat profile", "err", err)
-				}
+				return fmt.Errorf("failed to convert chat profile to streamplace chat profile: %w", err)
 			}
+			task.ChatProfile = spcp
+		}
 
-			for _, webhook := range atsync.CLI.DiscordWebhooks {
-				if webhook.DID == userDID && webhook.Type == "livestream" {
-					go func() {
-						err := discord.SendLivestream(ctx, webhook, r, lsv, postView, spcp)
-						if err != nil {
-							log.Error(ctx, "failed to send livestream to discord", "err", err)
-						} else {
-							log.Log(ctx, "sent livestream to discord", "user", userDID, "webhook", webhook.URL)
-						}
-					}()
-				}
+		if !isUpdate && !isFirstSync {
+			_, err = atsync.StatefulDB.EnqueueTask(ctx, statedb.TaskNotification, task, statedb.WithTaskKey(fmt.Sprintf("notification-blast::%s", aturi.String())))
+			if err != nil {
+				log.Error(ctx, "failed to enqueue notification task", "err", err)
 			}
 		}
 

pkg/cmd/streamplace.go

···
 	"syscall"
 	"time"
 
+	"github.com/bluesky-social/indigo/carstore"
 	"github.com/livepeer/go-livepeer/cmd/livepeer/starter"
 	"github.com/peterbourgon/ff/v3"
 	"github.com/streamplace/oatproxy/pkg/oatproxy"
···
 	"stream.place/streamplace/pkg/rtmps"
 	v0 "stream.place/streamplace/pkg/schema/v0"
 	"stream.place/streamplace/pkg/spmetrics"
+	"stream.place/streamplace/pkg/statedb"
 
 	"github.com/ThalesGroup/crypto11"
 	_ "github.com/go-gst/go-glib/glib"
···
 	if *version {
 		return nil
 	}
+
+	if len(os.Args) > 1 && os.Args[1] == "migrate" {
+		return statedb.Migrate(&cli)
+	}
+
 	spmetrics.Version.WithLabelValues(build.Version).Inc()
 	if cli.LivepeerHelp {
 		lpFlags := flag.NewFlagSet("livepeer", flag.ContinueOnError)
···
 		signer = hwsigner
 	}
 	var rep replication.Replicator = &boring.BoringReplicator{Peers: cli.Peers}
-	mod, err := model.MakeDB(cli.DBPath)
-	if err != nil {
-		return err
-	}
-	handle, err := atproto.MakeLexiconRepo(ctx, &cli, mod)
+
+	mod, err := model.MakeDB(cli.IndexDBPath)
 	if err != nil {
 		return err
 	}
-	defer handle.Close()
 	var noter notifications.FirebaseNotifier
 	if cli.FirebaseServiceAccount != "" {
 		noter, err = notifications.MakeFirebaseNotifier(ctx, cli.FirebaseServiceAccount)
···
 			return err
 		}
 	}
+	out := carstore.SQLiteStore{}
+	err = out.Open(":memory:")
+	if err != nil {
+		return err
+	}
+	state, err := statedb.MakeDB(&cli, noter, mod)
+	if err != nil {
+		return err
+	}
+	handle, err := atproto.MakeLexiconRepo(ctx, &cli, mod, state)
+	if err != nil {
+		return err
+	}
+	defer handle.Close()
 
-	jwkPath := cli.DataFilePath([]string{"jwk.json"})
-	jwk, err := atproto.EnsureJWK(ctx, jwkPath)
+	jwk, err := state.EnsureJWK(ctx, "jwk")
 	if err != nil {
 		return err
 	}
 	cli.JWK = jwk
 
-	accessJWKPath := cli.DataFilePath([]string{"access-jwk.json"})
-	accessJWK, err := atproto.EnsureJWK(ctx, accessJWKPath)
+	accessJWK, err := state.EnsureJWK(ctx, "access-jwk")
 	if err != nil {
 		return err
 	}
···
 
 	b := bus.NewBus()
 	atsync := &atproto.ATProtoSynchronizer{
-		CLI:   &cli,
-		Model: mod,
-		Noter: noter,
-		Bus:   b,
+		CLI:        &cli,
+		Model:      mod,
+		StatefulDB: state,
+		Noter:      noter,
+		Bus:        b,
 	}
 	mm, err := media.MakeMediaManager(ctx, &cli, signer, rep, mod, b, atsync)
 	if err != nil {
···
 
 	op := oatproxy.New(&oatproxy.Config{
 		Host:               cli.PublicHost,
-		CreateOAuthSession: mod.CreateOAuthSession,
-		UpdateOAuthSession: mod.UpdateOAuthSession,
-		GetOAuthSession:    mod.LoadOAuthSession,
+		CreateOAuthSession: state.CreateOAuthSession,
+		UpdateOAuthSession: state.UpdateOAuthSession,
+		GetOAuthSession:    state.LoadOAuthSession,
+		Lock:               state.GetNamedLock,
 		Scope:              "atproto transition:generic",
 		UpstreamJWK:        cli.JWK,
 		DownstreamJWK:      cli.AccessJWK,
 		ClientMetadata:     clientMetadata,
 	})
-	d := director.NewDirector(mm, mod, &cli, b, op)
-	a, err := api.MakeStreamplaceAPI(&cli, mod, eip712signer, noter, mm, ms, b, atsync, d, op)
+	d := director.NewDirector(mm, mod, &cli, b, op, state)
+	a, err := api.MakeStreamplaceAPI(&cli, mod, state, eip712signer, noter, mm, ms, b, atsync, d, op)
 	if err != nil {
 		return err
 	}
···
 
 	group.Go(func() error {
 		return handleSignals(ctx)
+	})
+
+	group.Go(func() error {
+		return state.ProcessQueue(ctx)
 	})
 
 	if cli.TracingEndpoint != "" {

pkg/config/config.go

···
 	AdminAccount           string
 	Build                  *BuildFlags
 	DataDir                string
-	DBPath                 string
+	DBURL                  string
+	IndexDBPath            string
 	EthAccountAddr         string
 	EthKeystorePath        string
 	EthPassword            string
···
 	cli.DataDirFlag(fs, &cli.TLSCertPath, "tls-cert", filepath.Join("tls", "tls.crt"), "Path to TLS certificate")
 	cli.DataDirFlag(fs, &cli.TLSKeyPath, "tls-key", filepath.Join("tls", "tls.key"), "Path to TLS key")
 	fs.StringVar(&cli.SigningKeyPath, "signing-key", "", "Path to signing key for pushing OTA updates to the app")
-	cli.DataDirFlag(fs, &cli.DBPath, "db-path", "db.sqlite", "path to sqlite database file")
+	fs.StringVar(&cli.DBURL, "db-url", "sqlite://$SP_DATA_DIR/state.sqlite", "URL of the database to use for storing private streamplace state")
+	cli.dataDirFlags = append(cli.dataDirFlags, &cli.DBURL)
+	cli.DataDirFlag(fs, &cli.IndexDBPath, "index-db-path", "db.sqlite", "path to sqlite database file for maintaining atproto index")
 	fs.StringVar(&cli.AdminAccount, "admin-account", "", "ethereum account that administrates this streamplace node")
 	fs.StringVar(&cli.FirebaseServiceAccount, "firebase-service-account", "", "JSON string of a firebase service account key")
 	fs.StringVar(&cli.GitLabURL, "gitlab-url", "https://git.stream.place/api/v4/projects/1", "gitlab url for generating download links")

pkg/director/director.go

···
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/media"
 	"stream.place/streamplace/pkg/model"
+	"stream.place/streamplace/pkg/statedb"
 )
 
 // director is responsible for managing the lifecycle of a stream, making business
···
 	streamSessions   map[string]*StreamSession
 	streamSessionsMu sync.Mutex
 	op               *oatproxy.OATProxy
+	statefulDB       *statedb.StatefulDB
 }
 
-func NewDirector(mm *media.MediaManager, mod model.Model, cli *config.CLI, bus *bus.Bus, op *oatproxy.OATProxy) *Director {
+func NewDirector(mm *media.MediaManager, mod model.Model, cli *config.CLI, bus *bus.Bus, op *oatproxy.OATProxy, statefulDB *statedb.StatefulDB) *Director {
 	return &Director{
 		mm:               mm,
 		mod:              mod,
···
 		streamSessions:   make(map[string]*StreamSession),
 		streamSessionsMu: sync.Mutex{},
 		op:               op,
+		statefulDB:       statefulDB,
 	}
 }
 
···
 					op:          d.op,
 					packets:     make([]bus.PacketizedSegment, 0),
 					started:     make(chan struct{}),
+					statefulDB:  d.statefulDB,
 				}
 				d.streamSessions[not.Segment.RepoDID] = ss
 				g.Go(func() error {

pkg/director/stream_session.go

···
 	"stream.place/streamplace/pkg/model"
 	"stream.place/streamplace/pkg/renditions"
 	"stream.place/streamplace/pkg/spmetrics"
+	"stream.place/streamplace/pkg/statedb"
 	"stream.place/streamplace/pkg/streamplace"
 	"stream.place/streamplace/pkg/thumbnail"
 )
···
 	started        chan struct{}
 	ctx            context.Context
 	packets        []bus.PacketizedSegment
+	statefulDB     *statedb.StatefulDB
 }
 
 func (ss *StreamSession) Start(ctx context.Context, not *media.NewSegmentNotification) error {
···
 		return nil
 	}
 
-	session, err := ss.mod.GetSessionByDID(repoDID)
+	session, err := ss.statefulDB.GetSessionByDID(repoDID)
 	if err != nil {
 		return fmt.Errorf("could not get OAuth session for repoDID: %w", err)
 	}

pkg/integrations/discord/avatars.go

···
 
 import (
 	"context"
-	"fmt"
 	"sync"
 
 	"github.com/bluesky-social/indigo/api/bsky"
 	"github.com/bluesky-social/indigo/xrpc"
 	"stream.place/streamplace/pkg/aqhttp"
-	"stream.place/streamplace/pkg/model"
 )
 
 var avatarCache = make(map[string]string)
···
 // getAvatarURL gets the avatar URL for a Bluesky from the public appview
 // pretty ugly. we're going to replace this with indexing bluesky profiles
 // at some point.
-func getAvatarURL(ctx context.Context, r *model.Repo) (string, error) {
+func getAvatarURL(ctx context.Context, did string) (string, error) {
 	avatarCacheMutex.Lock()
 	defer avatarCacheMutex.Unlock()
 
-	if r == nil || r.DID == "" {
-		return "", fmt.Errorf("repo or DID is nil or empty")
-	}
-
-	if avatar, ok := avatarCache[r.DID]; ok {
+	if avatar, ok := avatarCache[did]; ok {
 		return avatar, nil
 	}
 
···
 		Client: &aqhttp.Client,
 	}
 
-	profile, err := bsky.ActorGetProfile(ctx, xrpc, r.DID)
+	profile, err := bsky.ActorGetProfile(ctx, xrpc, did)
 	if err != nil {
 		return "", err
 	}
 
 	if profile.Avatar != nil {
-		avatarCache[r.DID] = *profile.Avatar
+		avatarCache[did] = *profile.Avatar
 		return *profile.Avatar, nil
 	}
 

pkg/integrations/discord/send-chat.go

···
 	"stream.place/streamplace/pkg/aqhttp"
 	"stream.place/streamplace/pkg/integrations/discord/discordtypes"
 	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/model"
 	"stream.place/streamplace/pkg/streamplace"
 )
 
-func SendChat(ctx context.Context, w *discordtypes.Webhook, r *model.Repo, scm *streamplace.ChatDefs_MessageView) error {
+func SendChat(ctx context.Context, w *discordtypes.Webhook, did string, scm *streamplace.ChatDefs_MessageView) error {
 
 	msg, ok := scm.Record.Val.(*streamplace.ChatMessage)
 	if !ok {
 		return fmt.Errorf("failed to cast chat message to streamplace chat message")
 	}
 
-	avatarURL, err := getAvatarURL(ctx, r)
+	avatarURL, err := getAvatarURL(ctx, did)
 	if err != nil {
 		log.Warn(ctx, "failed to get avatar URL", "err", err)
 	}

pkg/integrations/discord/send-livestream.go

···
 	"stream.place/streamplace/pkg/streamplace"
 )
 
-func SendLivestream(ctx context.Context, w *discordtypes.Webhook, r *model.Repo, lsv *streamplace.Livestream_LivestreamView, postView *bsky.FeedDefs_PostView, spcp *streamplace.ChatProfile) error {
+func SendLivestream(ctx context.Context, w *discordtypes.Webhook, pdsURL string, lsv *streamplace.Livestream_LivestreamView, postView *bsky.FeedDefs_PostView, spcp *streamplace.ChatProfile) error {
 	ctx = log.WithLogValues(ctx, "func", "SendLivestream")
 	ls, ok := lsv.Record.Val.(*streamplace.Livestream)
 	if !ok {
···
 	}
 
 	payload := discordtypes.Payload{
-		Username: fmt.Sprintf("@%s", r.Handle),
+		Username: fmt.Sprintf("@%s", lsv.Author.Handle),
 		Content:  fmt.Sprintf("%s%s%s", w.Prefix, content, w.Suffix),
 	}
 
-	avatarURL, err := getAvatarURL(ctx, r)
+	avatarURL, err := getAvatarURL(ctx, lsv.Author.Did)
 	if err != nil {
 		log.Warn(ctx, "failed to get avatar URL", "err", err)
 	}
···
 			log.Warn(ctx, "failed to parse URL", "err", err)
 		} else {
 			suffix = fmt.Sprintf(" on %s!", u.Host)
-			payload.Embeds[0].URL = fmt.Sprintf("%s/%s", *ls.Url, r.Handle)
+			payload.Embeds[0].URL = fmt.Sprintf("%s/%s", *ls.Url, lsv.Author.Handle)
 		}
 	}
 
-	payload.Embeds[0].Title = fmt.Sprintf("@%s is LIVE%s", r.Handle, suffix)
+	payload.Embeds[0].Title = fmt.Sprintf("@%s is LIVE%s", lsv.Author.Handle, suffix)
 
 	if ls.Thumb != nil {
-		u, err := url.Parse(fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob", r.PDS))
+		u, err := url.Parse(fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob", pdsURL))
 		if err != nil {
 			return fmt.Errorf("failed to parse base URL: %w", err)
 		}
 		q := u.Query()
-		q.Set("did", r.DID)
+		q.Set("did", lsv.Author.Did)
 		q.Set("cid", ls.Thumb.Ref.String())
 		u.RawQuery = q.Encode()
 		imageURL := u.String()

pkg/media/media_test.go

···
 		AllowedStreams: []string{"did:key:zQ3shhoPCrDZWE8CryCEHYCrb1x8mCkr2byTkF5EGJT7dgazC"},
 	})
 	atsync := &atproto.ATProtoSynchronizer{
-		CLI:   cli,
-		Model: mod,
-		Bus:   bus.NewBus(),
+		CLI:        cli,
+		Model:      mod,
+		StatefulDB: nil, // Test doesn't need StatefulDB for now
+		Bus:        bus.NewBus(),
 	}
 	mm, err := MakeMediaManager(context.Background(), cli, signer, &boring.BoringReplicator{}, mod, bus.NewBus(), atsync)
 	require.NoError(t, err)

pkg/model/model.go

···
 	"github.com/bluesky-social/indigo/api/bsky"
 	"github.com/lmittmann/tint"
 	slogGorm "github.com/orandin/slog-gorm"
-	"github.com/streamplace/oatproxy/pkg/oatproxy"
 	"gorm.io/driver/sqlite"
 	"gorm.io/gorm"
 	"stream.place/streamplace/pkg/config"
···
 }
 
 type Model interface {
-	CreateNotification(token, repoDID string) error
-	ListNotifications() ([]Notification, error)
-
 	CreatePlayerEvent(event PlayerEventAPI) error
 	ListPlayerEvents(playerID string) ([]PlayerEvent, error)
 	PlayerReport(playerID string) (map[string]any, error)
···
 	GetUserFollowers(ctx context.Context, userDID string) ([]Follow, error)
 	GetUserFollowingUser(ctx context.Context, userDID, subjectDID string) (*Follow, error)
 	DeleteFollow(ctx context.Context, userDID, rev string) error
-	GetFollowersNotificationTokens(userDID string) ([]string, error)
 
 	CreateFeedPost(ctx context.Context, post *FeedPost) error
 	ListFeedPosts() ([]FeedPost, error)
···
 	CreateChatProfile(ctx context.Context, profile *ChatProfile) error
 	GetChatProfile(ctx context.Context, repoDID string) (*ChatProfile, error)
 
-	CreateOAuthSession(id string, session *oatproxy.OAuthSession) error
-	LoadOAuthSession(id string) (*oatproxy.OAuthSession, error)
-	UpdateOAuthSession(id string, session *oatproxy.OAuthSession) error
-	ListOAuthSessions() ([]oatproxy.OAuthSession, error)
-	GetSessionByDID(did string) (*oatproxy.OAuthSession, error)
-
 	UpdateServerSettings(ctx context.Context, settings *ServerSettings) error
 	GetServerSettings(ctx context.Context, server string, repoDID string) (*ServerSettings, error)
 	DeleteServerSettings(ctx context.Context, server string, repoDID string) error
-
-	CreateCommitEvent(commit *comatproto.SyncSubscribeRepos_Commit, signedData string) error
-	GetCommitEventsSince(repoDID string, t time.Time) ([]*XrpcStreamEvent, error)
-	GetCommitEventsSinceSeq(repoDID string, seq int64) ([]*XrpcStreamEvent, error)
-	GetMostRecentCommitEvent(repoDID string) (*XrpcStreamEvent, error)
 
 	CreateLabeler(did string) (*Labeler, error)
 	GetLabeler(did string) (*Labeler, error)
···
 	}
 	sqlDB.SetMaxOpenConns(1)
 	for _, model := range []any{
-		Notification{},
 		PlayerEvent{},
 		Segment{},
 		Thumbnail{},
···
 		ChatMessage{},
 		ChatProfile{},
 		Gate{},
-		oatproxy.OAuthSession{},
 		ServerSettings{},
-		XrpcStreamEvent{},
+
 		Labeler{},
 		Label{},
 	} {

pkg/model/notification.go

···
-package model
-
-import (
-	"fmt"
-	"time"
-
-	"gorm.io/gorm"
-)
-
-type Notification struct {
-	Token     string `gorm:"primarykey"`
-	RepoDID   string `json:"repoDID,omitempty" gorm:"column:repo_did;index"`
-	CreatedAt time.Time
-	UpdatedAt time.Time
-	DeletedAt gorm.DeletedAt `gorm:"index"`
-}
-
-func (m *DBModel) CreateNotification(token string, repoDID string) error {
-	not := Notification{
-		Token: token,
-	}
-	if repoDID != "" {
-		not.RepoDID = repoDID
-	}
-	err := m.DB.Save(&not).Error
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
-func (m *DBModel) ListNotifications() ([]Notification, error) {
-	nots := []Notification{}
-	err := m.DB.Find(&nots).Error
-	if err != nil {
-		return nil, fmt.Errorf("error retrieving notifications: %w", err)
-	}
-	return nots, nil
-}
-
-func (m *DBModel) ListUserNotifications(userDID string) ([]Notification, error) {
-	nots := []Notification{}
-	err := m.DB.Where("repo_did = ?", userDID).Find(&nots).Error
-	if err != nil {
-		return nil, fmt.Errorf("error retrieving notifications: %w", err)
-	}
-	return nots, nil
-}
-
-func (m *DBModel) GetFollowersNotificationTokens(userDID string) ([]string, error) {
-	var tokens []string
-
-	err := m.DB.Model(&Notification{}).
-		Distinct("notifications.token").
-		Joins("JOIN follows ON follows.user_did = notifications.repo_did").
-		Where("follows.subject_did = ?", userDID).
-		Pluck("notifications.token", &tokens).
-		Error
-
-	if err != nil {
-		return nil, fmt.Errorf("error retrieving follower notification tokens: %w", err)
-	}
-
-	// also you prolly wanna get one for yourself
-	nots, err := m.ListUserNotifications(userDID)
-	if err != nil {
-		return nil, fmt.Errorf("error retrieving user notifications: %w", err)
-	}
-	for _, not := range nots {
-		tokens = append(tokens, not.Token)
-	}
-
-	return tokens, nil
-}

pkg/model/oauth_session.go

···
-package model
-
-import (
-	"errors"
-
-	"github.com/streamplace/oatproxy/pkg/oatproxy"
-	"gorm.io/gorm"
-)
-
-func (m *DBModel) CreateOAuthSession(id string, session *oatproxy.OAuthSession) error {
-	return m.DB.Create(session).Error
-}
-
-func (m *DBModel) LoadOAuthSession(id string) (*oatproxy.OAuthSession, error) {
-	var session oatproxy.OAuthSession
-	if err := m.DB.Where("downstream_dpop_jkt = ?", id).First(&session).Error; err != nil {
-		if errors.Is(err, gorm.ErrRecordNotFound) {
-			return nil, nil
-		}
-		return nil, err
-	}
-	return &session, nil
-}
-
-func (m *DBModel) UpdateOAuthSession(id string, session *oatproxy.OAuthSession) error {
-	res := m.DB.Model(&oatproxy.OAuthSession{}).Where("downstream_dpop_jkt = ?", id).Updates(session)
-	if res.Error != nil {
-		return res.Error
-	}
-	if res.RowsAffected == 0 {
-		return errors.New("no rows affected")
-	}
-	return nil
-}
-
-func (m *DBModel) ListOAuthSessions() ([]oatproxy.OAuthSession, error) {
-	var sessions []oatproxy.OAuthSession
-	if err := m.DB.Find(&sessions).Error; err != nil {
-		return nil, err
-	}
-	return sessions, nil
-}
-
-func (m *DBModel) GetSessionByDID(did string) (*oatproxy.OAuthSession, error) {
-	var session oatproxy.OAuthSession
-	if err := m.DB.Where("repo_did = ? AND revoked_at IS NULL", did).Order("updated_at DESC").First(&session).Error; err != nil {
-		return nil, err
-	}
-	return &session, nil
-}

pkg/model/xrpc_stream_event.go pkg/statedb/xrpc_stream_event.go

···
-package model
+package statedb
 
 import (
 	"bytes"
···
 )
 
 type XrpcStreamEvent struct {
-	CID        string    `json:"cid" gorm:"primaryKey"`
+	CID        string    `json:"cid" gorm:"column:cid;primaryKey"`
 	RepoDID    string    `json:"repoDID" gorm:"index:idx_repo_timestamp,priority:1;index:idx_repo_seq,priority:1;column:repo_did"`
 	Timestamp  time.Time `json:"timestamp" gorm:"index:idx_repo_timestamp,priority:2;column:timestamp"`
-	Data       []byte    `json:"data"`
+	Data       []byte    `json:"data" gorm:"column:data"`
 	SignedData string    `json:"signedData" gorm:"column:signed_data"`
 	Seq        int64     `json:"seq" gorm:"index:idx_repo_seq,priority:2;column:seq"`
 }
···
 	return commit, nil
 }
 
-func (m *DBModel) CreateCommitEvent(commit *comatproto.SyncSubscribeRepos_Commit, signedData string) error {
-	prev, err := m.GetMostRecentCommitEvent(commit.Repo)
+func (state *StatefulDB) CreateCommitEvent(commit *comatproto.SyncSubscribeRepos_Commit, signedData string) error {
+	prev, err := state.GetMostRecentCommitEvent(commit.Repo)
 	if err != nil {
 		return err
 	}
···
 		Seq:        commit.Seq,
 		SignedData: signedData,
 	}
-	return m.DB.Create(event).Error
+	return state.DB.Create(event).Error
 }
 
-func (m *DBModel) GetCommitEventsSince(repoDID string, t time.Time) ([]*XrpcStreamEvent, error) {
+func (state *StatefulDB) GetCommitEventsSince(repoDID string, t time.Time) ([]*XrpcStreamEvent, error) {
 	var events []*XrpcStreamEvent
-	query := m.DB.Where("repo_did = ?", repoDID)
+	query := state.DB.Where("repo_did = ?", repoDID)
 	query = query.Where("timestamp > ?", t.UTC())
 	err := query.Order("timestamp ASC").Find(&events).Error
 	if err != nil {
···
 	return events, nil
 }
 
-func (m *DBModel) GetCommitEventsSinceSeq(repoDID string, seq int64) ([]*XrpcStreamEvent, error) {
+func (state *StatefulDB) GetCommitEventsSinceSeq(repoDID string, seq int64) ([]*XrpcStreamEvent, error) {
 	var events []*XrpcStreamEvent
-	query := m.DB.Where("repo_did = ?", repoDID)
+	query := state.DB.Where("repo_did = ?", repoDID)
 	query = query.Where("seq > ?", seq)
 	err := query.Order("timestamp ASC").Find(&events).Error
 	if err != nil {
···
 	return events, nil
 }
 
-func (m *DBModel) GetMostRecentCommitEvent(repoDID string) (*XrpcStreamEvent, error) {
+func (state *StatefulDB) GetMostRecentCommitEvent(repoDID string) (*XrpcStreamEvent, error) {
 	var event XrpcStreamEvent
-	err := m.DB.Where("repo_did = ?", repoDID).
+	err := state.DB.Where("repo_did = ?", repoDID).
 		Order("timestamp DESC").
 		Limit(1).
 		First(&event).Error

pkg/resync/resync.go

···
 // resync a fresh database from the PDSses, copying over the few pieces of local state
 // that we have
 func Resync(ctx context.Context, cli *config.CLI) error {
-	oldMod, err := model.MakeDB(cli.DBPath)
+	oldMod, err := model.MakeDB(cli.IndexDBPath)
 	if err != nil {
 		return err
 	}
-	tempDBPath := cli.DBPath + ".temp." + fmt.Sprintf("%d", time.Now().UnixNano())
+	tempDBPath := cli.IndexDBPath + ".temp." + fmt.Sprintf("%d", time.Now().UnixNano())
 	newMod, err := model.MakeDB(tempDBPath)
 	if err != nil {
 		return err
···
 	}
 
 	atsync := &atproto.ATProtoSynchronizer{
-		CLI:   cli,
-		Model: newMod,
-		Noter: nil,
-		Bus:   bus.NewBus(),
+		CLI:        cli,
+		Model:      newMod,
+		StatefulDB: nil, // TODO: Add StatefulDB for resync when migration is ready
+		Noter:      nil,
+		Bus:        bus.NewBus(),
 	}
 
 	doneMap := make(map[string]bool)
···
 		return err
 	}
 
-	oauthSessions, err := oldMod.ListOAuthSessions()
-	if err != nil {
-		return err
-	}
-	for _, session := range oauthSessions {
-		err := newMod.CreateOAuthSession(session.DownstreamDPoPJKT, &session)
-		if err != nil {
-			return fmt.Errorf("failed to create oauth session: %w", err)
-		}
-	}
-	log.Log(ctx, "migrated oauth sessions", "count", len(oauthSessions))
+	// TODO: Update OAuth session migration to use new statefulDB
+	// oauthSessions, err := oldMod.ListOAuthSessions()
+	// if err != nil {
+	// 	return err
+	// }
+	// for _, session := range oauthSessions {
+	// 	err := newMod.CreateOAuthSession(session.DownstreamDPoPJKT, &session)
+	// 	if err != nil {
+	// 		return fmt.Errorf("failed to create oauth session: %w", err)
+	// 	}
+	// }
+	// log.Log(ctx, "migrated oauth sessions", "count", len(oauthSessions))
 
-	notificationTokens, err := oldMod.ListNotifications()
-	if err != nil {
-		return err
-	}
-	for _, token := range notificationTokens {
-		err := newMod.CreateNotification(token.Token, token.RepoDID)
-		if err != nil {
-			return fmt.Errorf("failed to create notification: %w", err)
-		}
-	}
-	log.Log(ctx, "migrated notification tokens", "count", len(notificationTokens))
+	// TODO: Update notification migration to use new statefulDB
+	// notificationTokens, err := oldMod.ListNotifications()
+	// if err != nil {
+	// 	return err
+	// }
+	// for _, token := range notificationTokens {
+	// 	err := newMod.CreateNotification(token.Token, token.RepoDID)
+	// 	if err != nil {
+	// 		return fmt.Errorf("failed to create notification: %w", err)
+	// 	}
+	// }
+	// log.Log(ctx, "migrated notification tokens", "count", len(notificationTokens))
 
 	log.Log(ctx, "resync complete!", "newDBPath", tempDBPath)
 

pkg/spxrpc/com_atproto_sync.go

···
 		return nil, fmt.Errorf("handleComAtprotoRepoGetRecord: failed to get user repo head: %w", err)
 	}
 
+	log.Warn(ctx, "got root", "root", root.String())
+
 	r, err := repo.OpenRepo(ctx, bs, root)
 	if err != nil {
 		return nil, fmt.Errorf("handleComAtprotoRepoGetRecord: failed to open repo: %w", err)
···
 	}
 
 	for _, blk := range blocks {
+		log.Warn(ctx, "writing block", "cid", blk.Cid().String(), "version", blk.Cid().Version())
 		if _, err := carstore.LdWrite(buf, blk.Cid().Bytes(), blk.RawData()); err != nil {
 			return nil, err
 		}
···
 		return err
 	}
 
-	evts, err := s.model.GetCommitEventsSinceSeq(atproto.LexiconRepo.RepoDid(), int64(seq))
+	evts, err := s.statefulDB.GetCommitEventsSinceSeq(atproto.LexiconRepo.RepoDid(), int64(seq))
 	if err != nil {
 		return err
 	}

pkg/spxrpc/spxrpc.go

···
 	"stream.place/streamplace/pkg/config"
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/model"
+	"stream.place/streamplace/pkg/statedb"
 )
 
 type Server struct {
···
 	model        model.Model
 	OGImageCache *cache.Cache
 	ATSync       *atproto.ATProtoSynchronizer
+	statefulDB   *statedb.StatefulDB
 }
 
-func NewServer(ctx context.Context, cli *config.CLI, model model.Model, op *oatproxy.OATProxy, mdlw middleware.Middleware, atsync *atproto.ATProtoSynchronizer) (*Server, error) {
+func NewServer(ctx context.Context, cli *config.CLI, model model.Model, statefulDB *statedb.StatefulDB, op *oatproxy.OATProxy, mdlw middleware.Middleware, atsync *atproto.ATProtoSynchronizer) (*Server, error) {
 	e := echo.New()
 	s := &Server{
 		e:            e,
···
 		model:        model,
 		OGImageCache: cache.New(5*time.Minute, 10*time.Minute), // 5min TTL, 10min cleanup
 		ATSync:       atsync,
+		statefulDB:   statefulDB,
 	}
 	e.Use(s.ErrorHandlingMiddleware())
 	e.Use(s.ContextPreservingMiddleware())

pkg/statedb/config.go

···
+package statedb
+
+import (
+	"errors"
+	"time"
+
+	"gorm.io/gorm"
+)
+
+type Config struct {
+	Key       string    `gorm:"column:key;primarykey"`
+	Value     []byte    `gorm:"column:value"`
+	CreatedAt time.Time `gorm:"column:created_at"`
+	UpdatedAt time.Time `gorm:"column:updated_at"`
+}
+
+func (state *StatefulDB) GetConfig(key string) (*Config, error) {
+	var config Config
+	if err := state.DB.Where("key = ?", key).First(&config).Error; err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &config, nil
+}
+
+func (state *StatefulDB) PutConfig(key string, value []byte) error {
+	config := Config{
+		Key:   key,
+		Value: value,
+	}
+	return state.DB.Save(&config).Error
+}

pkg/statedb/jwks.go

···
+package statedb
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"github.com/lestrrat-go/jwx/v2/jwk"
+	oauth_helpers "github.com/streamplace/atproto-oauth-golang/helpers"
+	"stream.place/streamplace/pkg/log"
+)
+
+func (state *StatefulDB) EnsureJWK(ctx context.Context, name string) (jwk.Key, error) {
+	var key jwk.Key
+
+	conf, err := state.GetConfig(name)
+	if err != nil {
+		return nil, err
+	}
+
+	// happy path: we found the jwk in the database, use that
+	if conf != nil {
+		key, err = jwk.ParseKey(conf.Value)
+		if err != nil {
+			return nil, err
+		}
+		return key, nil
+	}
+
+	// migration path: maybe we have an old one on disk.
+	key, _ = state.getOldJWK(ctx, name)
+
+	// new path: found neither, generate a new one
+	if key == nil {
+		log.Warn(ctx, "no JWK found, generating new one", "name", name)
+		key, err = oauth_helpers.GenerateKey(nil)
+		if err != nil {
+			return nil, fmt.Errorf("failed to generate JWK: %w", err)
+		}
+	}
+
+	b, err := json.Marshal(key)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal JWK: %w", err)
+	}
+	err = state.PutConfig(name, b)
+	if err != nil {
+		return nil, fmt.Errorf("failed to save JWK: %w", err)
+	}
+
+	return key, nil
+}
+
+// migration for the old one we stored on disk
+func (state *StatefulDB) getOldJWK(ctx context.Context, name string) (jwk.Key, error) {
+	var key jwk.Key
+	jwkPath := state.CLI.DataFilePath([]string{name + ".json"})
+	_, err := os.Stat(jwkPath)
+	if err == nil {
+		b, err := os.ReadFile(jwkPath)
+		if err != nil {
+			return nil, err
+		}
+		key, err = jwk.ParseKey(b)
+		if err != nil {
+			return nil, err
+		}
+		log.Warn(ctx, "found old JWK on disk, migrating to stateful database", "path", jwkPath)
+		return key, nil
+	}
+	return nil, nil
+}

pkg/statedb/locks.go

···
+package statedb
+
+import (
+	"crypto/sha256"
+	"encoding/binary"
+	"fmt"
+	"sync"
+)
+
+func (state *StatefulDB) GetNamedLock(name string) (func(), error) {
+	switch state.Type {
+	case DBTypeSQLite:
+		return state.getNamedLockSQLite(name)
+	case DBTypePostgres:
+		return state.getNamedLockPostgres(name)
+	}
+	panic("unsupported database type")
+}
+
+func (state *StatefulDB) getNamedLockPostgres(name string) (func(), error) {
+	// Convert string to sha256 hash and use decimal value for advisory lock
+	h := sha256.Sum256([]byte(name))
+	nameInt := binary.BigEndian.Uint64(h[:32])
+
+	err := state.DB.Exec("SELECT pg_advisory_lock($1)", nameInt).Error
+	if err != nil {
+		return nil, err
+	}
+	return func() {
+		err := state.DB.Exec("SELECT pg_advisory_unlock($1)", nameInt).Error
+		if err != nil {
+			// unfortunate, but the risk is that we're holding on to the lock forever,
+			// so it's responsible to crash in this case
+			panic(fmt.Errorf("error unlocking named lock: %w", err))
+		}
+	}, nil
+}
+
+func (state *StatefulDB) getNamedLockSQLite(name string) (func(), error) {
+	lock := state.locks.GetLock(name)
+	lock.Lock()
+	return func() {
+		lock.Unlock()
+	}, nil
+}
+
+// Local mutex implementation for sqlite
+type NamedLocks struct {
+	mu    sync.Mutex
+	locks map[string]*sync.Mutex
+}
+
+// NewNamedLocks creates a new NamedLocks instance
+func NewNamedLocks() *NamedLocks {
+	return &NamedLocks{
+		locks: make(map[string]*sync.Mutex),
+	}
+}
+
+// GetLock returns the mutex for the given name, creating it if it doesn't exist
+func (n *NamedLocks) GetLock(name string) *sync.Mutex {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+
+	lock, exists := n.locks[name]
+	if !exists {
+		lock = &sync.Mutex{}
+		n.locks[name] = lock
+	}
+	return lock
+}

pkg/statedb/migrate.go

···
+package statedb
+
+import (
+	"context"
+	"os"
+	"time"
+
+	"github.com/lmittmann/tint"
+	slogGorm "github.com/orandin/slog-gorm"
+	"github.com/streamplace/oatproxy/pkg/oatproxy"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+	"stream.place/streamplace/pkg/config"
+	"stream.place/streamplace/pkg/log"
+)
+
+func Migrate(cli *config.CLI) error {
+	gormLogger := slogGorm.New(
+		slogGorm.WithHandler(tint.NewHandler(os.Stderr, &tint.Options{
+			TimeFormat: time.RFC3339,
+		})),
+		// slogGorm.WithTraceAll(),
+	)
+
+	newDB, err := MakeDB(cli, nil, nil)
+	if err != nil {
+		return err
+	}
+
+	oldDB, err := gorm.Open(sqlite.Open(cli.IndexDBPath), &gorm.Config{
+		Logger: gormLogger,
+	})
+	if err != nil {
+		return err
+	}
+
+	var sessions []oatproxy.OAuthSession
+	if err := oldDB.Find(&sessions).Error; err != nil {
+		return err
+	}
+
+	for _, session := range sessions {
+		log.Log(context.Background(), "migrating session", "session", session.DownstreamDPoPJKT)
+		err := newDB.DB.Save(&session).Error
+		if err != nil {
+			return err
+		}
+	}
+
+	var notifications []Notification
+	if err := oldDB.Find(&notifications).Error; err != nil {
+		time.Sleep(1 * time.Second)
+		return err
+	}
+
+	for _, notification := range notifications {
+		log.Log(context.Background(), "migrating notification", "notification", notification)
+		err := newDB.DB.Save(&notification).Error
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

pkg/statedb/notification.go

···
+package statedb
+
+import (
+	"fmt"
+	"time"
+)
+
+type Notification struct {
+	Token     string    `gorm:"column:token;primarykey"`
+	RepoDID   string    `json:"repoDID,omitempty" gorm:"column:repo_did;index"`
+	CreatedAt time.Time `gorm:"column:created_at"`
+	UpdatedAt time.Time `gorm:"column:updated_at"`
+}
+
+func (state *StatefulDB) CreateNotification(token string, repoDID string) error {
+	not := Notification{
+		Token: token,
+	}
+	if repoDID != "" {
+		not.RepoDID = repoDID
+	}
+	err := state.DB.Save(&not).Error
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (state *StatefulDB) ListNotifications() ([]Notification, error) {
+	nots := []Notification{}
+	err := state.DB.Find(&nots).Error
+	if err != nil {
+		return nil, fmt.Errorf("error retrieving notifications: %w", err)
+	}
+	return nots, nil
+}
+
+func (state *StatefulDB) ListUserNotifications(userDID string) ([]Notification, error) {
+	nots := []Notification{}
+	err := state.DB.Where("repo_did = ?", userDID).Find(&nots).Error
+	if err != nil {
+		return nil, fmt.Errorf("error retrieving notifications: %w", err)
+	}
+	return nots, nil
+}
+
+func (state *StatefulDB) GetManyNotificationTokens(userDIDs []string) ([]string, error) {
+	tokens := []string{}
+	err := state.DB.Model(&Notification{}).
+		Where("repo_did IN (?)", userDIDs).
+		Pluck("token", &tokens).
+		Error
+	if err != nil {
+		return nil, fmt.Errorf("error retrieving notification tokens: %w", err)
+	}
+	return tokens, nil
+}

pkg/statedb/oauth_session.go

···
+package statedb
+
+import (
+	"errors"
+
+	"github.com/streamplace/oatproxy/pkg/oatproxy"
+	"gorm.io/gorm"
+)
+
+func (state *StatefulDB) CreateOAuthSession(id string, session *oatproxy.OAuthSession) error {
+	return state.DB.Create(session).Error
+}
+
+func (state *StatefulDB) LoadOAuthSession(id string) (*oatproxy.OAuthSession, error) {
+	var session oatproxy.OAuthSession
+	if err := state.DB.Where("downstream_dpop_jkt = ?", id).First(&session).Error; err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &session, nil
+}
+
+func (state *StatefulDB) UpdateOAuthSession(id string, session *oatproxy.OAuthSession) error {
+	res := state.DB.Model(&oatproxy.OAuthSession{}).Where("downstream_dpop_jkt = ?", id).Updates(session)
+	if res.Error != nil {
+		return res.Error
+	}
+	if res.RowsAffected == 0 {
+		return errors.New("no rows affected")
+	}
+	return nil
+}
+
+func (state *StatefulDB) ListOAuthSessions() ([]oatproxy.OAuthSession, error) {
+	var sessions []oatproxy.OAuthSession
+	if err := state.DB.Find(&sessions).Error; err != nil {
+		return nil, err
+	}
+	return sessions, nil
+}
+
+func (state *StatefulDB) GetSessionByDID(did string) (*oatproxy.OAuthSession, error) {
+	var session oatproxy.OAuthSession
+	if err := state.DB.Where("repo_did = ? AND revoked_at IS NULL", did).Order("updated_at DESC").First(&session).Error; err != nil {
+		return nil, err
+	}
+	return &session, nil
+}

pkg/statedb/queue_processor.go

···
+package statedb
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/bluesky-social/indigo/api/bsky"
+	"gorm.io/gorm"
+	"stream.place/streamplace/pkg/integrations/discord"
+	"stream.place/streamplace/pkg/log"
+	notificationpkg "stream.place/streamplace/pkg/notifications"
+	"stream.place/streamplace/pkg/streamplace"
+)
+
+var TaskNotification = "notification"
+var TaskChat = "chat"
+
+type NotificationTask struct {
+	Livestream  *streamplace.Livestream_LivestreamView
+	FeedPost    *bsky.FeedDefs_PostView
+	ChatProfile *streamplace.ChatProfile
+	PDSURL      string
+}
+
+type ChatTask struct {
+	MessageView *streamplace.ChatDefs_MessageView
+}
+
+func (state *StatefulDB) ProcessQueue(ctx context.Context) error {
+	for {
+		task, err := state.DequeueTask(ctx, "queue_processor")
+		if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
+			return err
+		}
+		if task != nil {
+			err := state.processTask(ctx, task)
+			if err != nil {
+				log.Error(ctx, "failed to process task", "err", err)
+			}
+		} else {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-time.After(1 * time.Second):
+				continue
+			case <-state.pokeQueue:
+				continue
+			}
+		}
+
+	}
+}
+
+func (state *StatefulDB) processTask(ctx context.Context, task *AppTask) error {
+	switch task.Type {
+	case TaskNotification:
+		return state.processNotificationTask(ctx, task)
+	case TaskChat:
+		return state.processChatMessageTask(ctx, task)
+	default:
+		return fmt.Errorf("unknown task type: %s", task.Type)
+	}
+}
+
+func (state *StatefulDB) processNotificationTask(ctx context.Context, task *AppTask) error {
+	var notificationTask NotificationTask
+	if err := json.Unmarshal(task.Payload, &notificationTask); err != nil {
+		return err
+	}
+	lsv := notificationTask.Livestream
+	rec, ok := lsv.Record.Val.(*streamplace.Livestream)
+	if !ok {
+		return fmt.Errorf("invalid livestream record")
+	}
+	userDID := lsv.Author.Did
+
+	log.Warn(ctx, "Livestream detected! Blasting followers!", "title", rec.Title, "url", rec.Url, "createdAt", rec.CreatedAt, "repo", userDID)
+	followers, err := state.model.GetUserFollowers(ctx, userDID)
+	if err != nil {
+		return err
+	}
+
+	followersDIDs := make([]string, 0, len(followers))
+	for _, follower := range followers {
+		followersDIDs = append(followersDIDs, follower.UserDID)
+	}
+
+	log.Log(ctx, "found followers", "count", len(followersDIDs))
+
+	notifications, err := state.GetManyNotificationTokens(followersDIDs)
+	if err != nil {
+		return err
+	}
+
+	if state.noter != nil {
+		nb := &notificationpkg.NotificationBlast{
+			Title: fmt.Sprintf("🔴 @%s is LIVE!", lsv.Author.Handle),
+			Body:  rec.Title,
+			Data: map[string]string{
+				"path": fmt.Sprintf("/%s", lsv.Author.Handle),
+			},
+		}
+		err = state.noter.Blast(ctx, notifications, nb)
+		if err != nil {
+			log.Error(ctx, "failed to blast notifications", "err", err)
+		} else {
+			log.Log(ctx, "sent notifications", "user", userDID, "count", len(notifications), "content", nb)
+		}
+	} else {
+		log.Log(ctx, "no notifier configured, skipping notifications", "user", userDID, "count", len(notifications))
+	}
+
+	for _, webhook := range state.CLI.DiscordWebhooks {
+		if webhook.DID == userDID && webhook.Type == "livestream" {
+			go func() {
+				err := discord.SendLivestream(ctx, webhook, notificationTask.PDSURL, lsv, notificationTask.FeedPost, notificationTask.ChatProfile)
+				if err != nil {
+					log.Error(ctx, "failed to send livestream to discord", "err", err)
+				} else {
+					log.Log(ctx, "sent livestream to discord", "user", userDID, "webhook", webhook.URL)
+				}
+			}()
+		}
+	}
+	return nil
+}
+
+func (state *StatefulDB) processChatMessageTask(ctx context.Context, task *AppTask) error {
+	var chatTask ChatTask
+	if err := json.Unmarshal(task.Payload, &chatTask); err != nil {
+		return err
+	}
+	scm := chatTask.MessageView
+	rec, ok := scm.Record.Val.(*streamplace.ChatMessage)
+	if !ok {
+		return fmt.Errorf("invalid chat message record")
+	}
+	userDID := scm.Author.Did
+
+	for _, webhook := range state.CLI.DiscordWebhooks {
+		if webhook.DID == rec.Streamer && webhook.Type == "chat" {
+			go func() {
+				err := discord.SendChat(ctx, webhook, scm.Author.Did, scm)
+				if err != nil {
+					log.Error(ctx, "failed to send livestream to discord", "err", err)
+				} else {
+					log.Log(ctx, "sent livestream to discord", "user", userDID, "webhook", webhook.URL)
+				}
+			}()
+		}
+	}
+	return nil
+}

pkg/statedb/statedb.go

···
+package statedb
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/lmittmann/tint"
+	slogGorm "github.com/orandin/slog-gorm"
+	"github.com/streamplace/oatproxy/pkg/oatproxy"
+	"gorm.io/driver/postgres"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+	"stream.place/streamplace/pkg/config"
+	"stream.place/streamplace/pkg/log"
+	"stream.place/streamplace/pkg/model"
+	notificationpkg "stream.place/streamplace/pkg/notifications"
+)
+
+type DBType string
+
+const (
+	DBTypeSQLite   DBType = "sqlite"
+	DBTypePostgres DBType = "postgres"
+)
+
+type StatefulDB struct {
+	DB    *gorm.DB
+	CLI   *config.CLI
+	Type  DBType
+	locks *NamedLocks
+	noter notificationpkg.FirebaseNotifier
+	model model.Model
+	// pokeQueue is used to wake up the queue processor when a new task is enqueued
+	pokeQueue chan struct{}
+}
+
+// list tables here so we can migrate them
+var StatefulDBModels = []any{
+	oatproxy.OAuthSession{},
+	Notification{},
+	Config{},
+	XrpcStreamEvent{},
+	AppTask{},
+}
+
+var NoPostgresDatabaseCode = "3D000"
+
+// Stateful database for storing private streamplace state
+func MakeDB(cli *config.CLI, noter notificationpkg.FirebaseNotifier, model model.Model) (*StatefulDB, error) {
+	dbURL := cli.DBURL
+	log.Log(context.Background(), "starting stateful database", "dbURL", redactDBURL(dbURL))
+	var dial gorm.Dialector
+	var dbType DBType
+	if dbURL == ":memory:" {
+		dial = sqlite.Open(":memory:")
+		dbType = DBTypeSQLite
+	} else if strings.HasPrefix(dbURL, "sqlite://") {
+		dial = sqlite.Open(dbURL[len("sqlite://"):])
+		dbType = DBTypeSQLite
+	} else if strings.HasPrefix(dbURL, "postgres://") {
+		dial = postgres.Open(dbURL)
+		dbType = DBTypePostgres
+	} else {
+		return nil, fmt.Errorf("unsupported database URL (most start with sqlite:// or postgres://): %s", redactDBURL(dbURL))
+	}
+
+	db, err := openDB(dial)
+
+	if err != nil {
+		if dbType == DBTypePostgres && strings.Contains(err.Error(), NoPostgresDatabaseCode) {
+			db, err = makePostgresDB(dbURL)
+			if err != nil {
+				return nil, fmt.Errorf("error creating streamplace database: %w", err)
+			}
+		} else {
+			return nil, fmt.Errorf("error starting database: %w", err)
+		}
+	}
+	if dbType == DBTypeSQLite {
+		err = db.Exec("PRAGMA journal_mode=WAL;").Error
+		if err != nil {
+			return nil, fmt.Errorf("error setting journal mode: %w", err)
+		}
+		sqlDB, err := db.DB()
+		if err != nil {
+			return nil, fmt.Errorf("error getting database: %w", err)
+		}
+		sqlDB.SetMaxOpenConns(1)
+	}
+	for _, model := range StatefulDBModels {
+		err = db.AutoMigrate(model)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return &StatefulDB{
+		DB:        db,
+		CLI:       cli,
+		Type:      dbType,
+		locks:     NewNamedLocks(),
+		model:     model,
+		pokeQueue: make(chan struct{}, 1),
+	}, nil
+}
+
+func openDB(dial gorm.Dialector) (*gorm.DB, error) {
+	gormLogger := slogGorm.New(
+		slogGorm.WithHandler(tint.NewHandler(os.Stderr, &tint.Options{
+			TimeFormat: time.RFC3339,
+		})),
+		// slogGorm.WithTraceAll(),
+	)
+
+	return gorm.Open(dial, &gorm.Config{
+		SkipDefaultTransaction: true,
+		TranslateError:         true,
+		Logger:                 gormLogger,
+	})
+}
+
+// helper function for creating the requested postgres database
+func makePostgresDB(dbURL string) (*gorm.DB, error) {
+	u, err := url.Parse(dbURL)
+	if err != nil {
+		return nil, err
+	}
+	dbName := strings.TrimPrefix(u.Path, "/")
+	u.Path = "/postgres"
+
+	rootDial := postgres.Open(u.String())
+
+	db, err := openDB(rootDial)
+	if err != nil {
+		return nil, err
+	}
+
+	// postgres doesn't support prepared statements for CREATE DATABASE. don't SQL inject yourself.
+	err = db.Exec(fmt.Sprintf("CREATE DATABASE %s;", dbName)).Error
+	if err != nil {
+		return nil, err
+	}
+
+	log.Warn(context.Background(), "created postgres database", "dbName", dbName)
+
+	realDial := postgres.Open(dbURL)
+
+	return openDB(realDial)
+}
+
+func redactDBURL(dbURL string) string {
+	u, err := url.Parse(dbURL)
+	if err != nil {
+		return "db url is malformed"
+	}
+	if u.User != nil {
+		u.User = url.UserPassword(u.User.Username(), "redacted")
+	}
+	return u.String()
+}

pkg/statedb/task.go

···
+package statedb
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"gorm.io/gorm"
+)
+
+// TaskStatus represents the status of a task in the queue
+type TaskStatus string
+
+const (
+	TaskStatusPending    TaskStatus = "PENDING"
+	TaskStatusProcessing TaskStatus = "PROCESSING"
+	TaskStatusCompleted  TaskStatus = "COMPLETED"
+	TaskStatusFailed     TaskStatus = "FAILED"
+	TaskStatusRetrying   TaskStatus = "RETRYING"
+)
+
+// AppTask represents a task in the queue
+type AppTask struct {
+	ID          uint            `gorm:"column:id;primarykey"`
+	Type        string          `gorm:"column:type;not null;index"`
+	TaskKey     *string         `gorm:"column:task_key;index:idx_task_dedup,unique"`
+	Status      TaskStatus      `gorm:"column:status;not null;index;default:'PENDING'"`
+	Payload     json.RawMessage `gorm:"column:payload;type:jsonb"`
+	Priority    int             `gorm:"column:priority;default:0;index"`
+	TryCount    int             `gorm:"column:try_count;default:0"`
+	MaxTries    int             `gorm:"column:max_tries;default:3"`
+	LockExpires *time.Time      `gorm:"column:lock_expires"`
+	WorkerID    *string         `gorm:"column:worker_id"`
+	Error       *string         `gorm:"column:error"`
+	CreatedAt   time.Time       `gorm:"column:created_at"`
+	UpdatedAt   time.Time       `gorm:"column:updated_at"`
+	ScheduledAt *time.Time      `gorm:"column:scheduled_at"` // for delayed tasks
+}
+
+// EnqueueTask adds a new task to the queue
+func (state *StatefulDB) EnqueueTask(ctx context.Context, taskType string, payload any, options ...TaskOption) (*AppTask, error) {
+	payloadBytes, err := json.Marshal(payload)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal payload: %w", err)
+	}
+
+	task := &AppTask{
+		Type:     taskType,
+		Status:   TaskStatusPending,
+		Payload:  payloadBytes,
+		Priority: 0,
+		MaxTries: 3,
+	}
+
+	// Apply options
+	for _, opt := range options {
+		opt(task)
+	}
+
+	// If task has a key, check for deduplication
+	if task.TaskKey != nil {
+		existingTask, err := state.GetTaskByKey(ctx, *task.TaskKey)
+		if err != nil {
+			return nil, fmt.Errorf("failed to check for existing task: %w", err)
+		}
+		if existingTask != nil {
+			// Task already exists, return the existing one
+			return existingTask, nil
+		}
+	}
+
+	if err := state.DB.WithContext(ctx).Create(task).Error; err != nil {
+		// Handle unique constraint violation gracefully
+		if strings.Contains(err.Error(), "duplicate") || strings.Contains(err.Error(), "UNIQUE constraint") {
+			// Another node beat us to it, try to fetch the existing task
+			if task.TaskKey != nil {
+				existingTask, fetchErr := state.GetTaskByKey(ctx, *task.TaskKey)
+				if fetchErr == nil && existingTask != nil {
+					return existingTask, nil
+				}
+			}
+		}
+		return nil, fmt.Errorf("failed to enqueue task: %w", err)
+	}
+
+	go func() {
+		select {
+		case state.pokeQueue <- struct{}{}:
+			// wake up the queue processor
+		default:
+			// queue is already awake, do nothing
+		}
+	}()
+
+	return task, nil
+}
+
+// DequeueTask retrieves the next available task from the queue and locks it
+func (state *StatefulDB) DequeueTask(ctx context.Context, workerID string, taskTypes ...string) (*AppTask, error) {
+	var task AppTask
+
+	err := state.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		query := tx.Where("status = ?", TaskStatusPending).
+			Where("try_count < max_tries").
+			Where("(lock_expires IS NULL OR lock_expires < ?)", time.Now()).
+			Where("(scheduled_at IS NULL OR scheduled_at <= ?)", time.Now())
+
+		if len(taskTypes) > 0 {
+			query = query.Where("type IN ?", taskTypes)
+		}
+
+		// Use raw SQL for PostgreSQL-specific locking
+		if state.Type == DBTypePostgres {
+			baseQuery := "SELECT * FROM app_tasks WHERE status = ? AND try_count < max_tries AND (lock_expires IS NULL OR lock_expires < ?) AND (scheduled_at IS NULL OR scheduled_at <= ?)"
+			if len(taskTypes) > 0 {
+				baseQuery += " AND type IN ?"
+				params := []interface{}{TaskStatusPending, time.Now(), time.Now(), taskTypes}
+				err := tx.Raw(baseQuery+" ORDER BY priority DESC, created_at ASC LIMIT 1 FOR UPDATE SKIP LOCKED", params...).
+					Scan(&task).Error
+				if err != nil {
+					return err
+				}
+			} else {
+				err := tx.Raw(baseQuery+" ORDER BY priority DESC, created_at ASC LIMIT 1 FOR UPDATE SKIP LOCKED",
+					TaskStatusPending, time.Now(), time.Now()).
+					Scan(&task).Error
+				if err != nil {
+					return err
+				}
+			}
+		} else {
+			// Fallback for SQLite (no SKIP LOCKED support)
+			err := query.Order("priority DESC, created_at ASC").First(&task).Error
+			if err != nil {
+				return err
+			}
+		}
+
+		if task.ID == 0 {
+			return gorm.ErrRecordNotFound
+		}
+
+		// Lock the task
+		lockExpires := time.Now().Add(30 * time.Minute) // 30-minute lock
+		updates := map[string]interface{}{
+			"status":       TaskStatusProcessing,
+			"worker_id":    workerID,
+			"lock_expires": lockExpires,
+			"try_count":    task.TryCount + 1,
+		}
+
+		return tx.Model(&task).Updates(updates).Error
+	})
+
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, nil // No tasks available
+		}
+		return nil, fmt.Errorf("failed to dequeue task: %w", err)
+	}
+
+	// Reload the task to get updated fields
+	if err := state.DB.WithContext(ctx).First(&task, task.ID).Error; err != nil {
+		return nil, fmt.Errorf("failed to reload task: %w", err)
+	}
+
+	return &task, nil
+}
+
+// CompleteTask marks a task as completed
+func (state *StatefulDB) CompleteTask(ctx context.Context, taskID uint) error {
+	result := state.DB.WithContext(ctx).Model(&AppTask{}).
+		Where("id = ?", taskID).
+		Updates(map[string]interface{}{
+			"status":       TaskStatusCompleted,
+			"lock_expires": nil,
+			"worker_id":    nil,
+		})
+
+	if result.Error != nil {
+		return fmt.Errorf("failed to complete task: %w", result.Error)
+	}
+
+	if result.RowsAffected == 0 {
+		return errors.New("task not found")
+	}
+
+	return nil
+}
+
+// FailTask marks a task as failed and optionally retries it
+func (state *StatefulDB) FailTask(ctx context.Context, taskID uint, errorMsg string) error {
+	var task AppTask
+	err := state.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		if err := tx.First(&task, taskID).Error; err != nil {
+			return err
+		}
+
+		updates := map[string]interface{}{
+			"error":        errorMsg,
+			"lock_expires": nil,
+			"worker_id":    nil,
+		}
+
+		if task.TryCount >= task.MaxTries {
+			updates["status"] = TaskStatusFailed
+		} else {
+			updates["status"] = TaskStatusPending
+		}
+
+		return tx.Model(&task).Updates(updates).Error
+	})
+
+	if err != nil {
+		return fmt.Errorf("failed to mark task as failed: %w", err)
+	}
+
+	return nil
+}
+
+// ReleaseTask releases a locked task back to the queue (e.g., worker shutdown)
+func (state *StatefulDB) ReleaseTask(ctx context.Context, taskID uint) error {
+	result := state.DB.WithContext(ctx).Model(&AppTask{}).
+		Where("id = ?", taskID).
+		Updates(map[string]interface{}{
+			"status":       TaskStatusPending,
+			"lock_expires": nil,
+			"worker_id":    nil,
+		})
+
+	if result.Error != nil {
+		return fmt.Errorf("failed to release task: %w", result.Error)
+	}
+
+	if result.RowsAffected == 0 {
+		return errors.New("task not found")
+	}
+
+	return nil
+}
+
+// GetTask retrieves a task by ID
+func (state *StatefulDB) GetTask(ctx context.Context, taskID uint) (*AppTask, error) {
+	var task AppTask
+	if err := state.DB.WithContext(ctx).First(&task, taskID).Error; err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("failed to get task: %w", err)
+	}
+	return &task, nil
+}
+
+// GetTaskByKey retrieves a task by its unique task key
+func (state *StatefulDB) GetTaskByKey(ctx context.Context, taskKey string) (*AppTask, error) {
+	var task AppTask
+	if err := state.DB.WithContext(ctx).Where("task_key = ?", taskKey).First(&task).Error; err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("failed to get task by key: %w", err)
+	}
+	return &task, nil
+}
+
+// ListTasks retrieves tasks with optional filters
+func (state *StatefulDB) ListTasks(ctx context.Context, filters TaskFilters) ([]AppTask, error) {
+	var tasks []AppTask
+	query := state.DB.WithContext(ctx).Model(&AppTask{})
+
+	if filters.Status != "" {
+		query = query.Where("status = ?", filters.Status)
+	}
+	if filters.Type != "" {
+		query = query.Where("type = ?", filters.Type)
+	}
+	if filters.TaskKey != "" {
+		query = query.Where("task_key = ?", filters.TaskKey)
+	}
+	if filters.WorkerID != "" {
+		query = query.Where("worker_id = ?", filters.WorkerID)
+	}
+	if filters.Limit > 0 {
+		query = query.Limit(filters.Limit)
+	}
+	if filters.Offset > 0 {
+		query = query.Offset(filters.Offset)
+	}
+
+	query = query.Order("created_at DESC")
+
+	if err := query.Find(&tasks).Error; err != nil {
+		return nil, fmt.Errorf("failed to list tasks: %w", err)
+	}
+
+	return tasks, nil
+}
+
+// CleanupExpiredLocks releases tasks with expired locks
+func (state *StatefulDB) CleanupExpiredLocks(ctx context.Context) (int64, error) {
+	result := state.DB.WithContext(ctx).Model(&AppTask{}).
+		Where("status = ? AND lock_expires < ?", TaskStatusProcessing, time.Now()).
+		Updates(map[string]interface{}{
+			"status":       TaskStatusPending,
+			"lock_expires": nil,
+			"worker_id":    nil,
+		})
+
+	if result.Error != nil {
+		return 0, fmt.Errorf("failed to cleanup expired locks: %w", result.Error)
+	}
+
+	return result.RowsAffected, nil
+}
+
+// TaskOption is a function that configures a task
+type TaskOption func(*AppTask)
+
+// WithPriority sets the task priority (higher numbers = higher priority)
+func WithPriority(priority int) TaskOption {
+	return func(t *AppTask) {
+		t.Priority = priority
+	}
+}
+
+// WithMaxTries sets the maximum number of retry attempts
+func WithMaxTries(maxTries int) TaskOption {
+	return func(t *AppTask) {
+		t.MaxTries = maxTries
+	}
+}
+
+// WithScheduledAt sets when the task should be processed (for delayed tasks)
+func WithScheduledAt(scheduledAt time.Time) TaskOption {
+	return func(t *AppTask) {
+		t.ScheduledAt = &scheduledAt
+	}
+}
+
+// WithTaskKey sets a unique key for task deduplication
+func WithTaskKey(taskKey string) TaskOption {
+	return func(t *AppTask) {
+		t.TaskKey = &taskKey
+	}
+}
+
+// TaskFilters holds filters for listing tasks
+type TaskFilters struct {
+	Status   TaskStatus
+	Type     string
+	TaskKey  string
+	WorkerID string
+	Limit    int
+	Offset   int
+}