From 78d89a2bc613e95524b87b7ea6901f50f5cd756c Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Fri, 29 Aug 2025 18:27:32 -0700 Subject: [PATCH] model: implement reindexing, index things by URI instead of CID --- pkg/api/api_internal.go | 10 +-- pkg/atproto/atproto.go | 4 +- pkg/atproto/labeler_firehose.go | 4 +- pkg/atproto/locks.go | 47 +---------- pkg/atproto/migrate.go | 91 +++++++++++++++++++++ pkg/atproto/sync.go | 6 +- pkg/cmd/streamplace.go | 12 +-- pkg/config/config.go | 2 - pkg/model/chat_message.go | 8 +- pkg/model/feed_post.go | 14 ++-- pkg/model/livestream.go | 14 ++-- pkg/model/model.go | 21 ++--- pkg/resync/resync.go | 113 --------------------------- pkg/spxrpc/com_atproto_moderation.go | 2 +- pkg/statedb/migrate.go | 2 +- 15 files changed, 144 insertions(+), 206 deletions(-) create mode 100644 pkg/atproto/migrate.go delete mode 100644 pkg/resync/resync.go diff --git a/pkg/api/api_internal.go b/pkg/api/api_internal.go index 59a5a52c..169e7d3a 100644 --- a/pkg/api/api_internal.go +++ b/pkg/api/api_internal.go @@ -410,13 +410,13 @@ func (a *StreamplaceAPI) InternalHandler(ctx context.Context) (http.Handler, err } }) - router.GET("/chat/:cid", func(w http.ResponseWriter, r *http.Request, p httprouter.Params) { - cid := p.ByName("cid") - if cid == "" { - errors.WriteHTTPBadRequest(w, "cid required", nil) + router.GET("/chat/:uri", func(w http.ResponseWriter, r *http.Request, p httprouter.Params) { + uri := p.ByName("uri") + if uri == "" { + errors.WriteHTTPBadRequest(w, "uri required", nil) return } - msg, err := a.Model.GetChatMessage(cid) + msg, err := a.Model.GetChatMessage(uri) if err != nil { errors.WriteHTTPInternalServerError(w, "unable to get chat posts", err) return diff --git a/pkg/atproto/atproto.go b/pkg/atproto/atproto.go index 62434a3f..d52a4fd8 100644 --- a/pkg/atproto/atproto.go +++ b/pkg/atproto/atproto.go @@ -49,7 +49,7 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle s ctx = log.WithLogValues(ctx, "did", ident.DID.String(), "handle", ident.Handle.String()) - handleLock := getHandleLock(ident.DID.String()) + handleLock := handleLocks.GetLock(ident.DID.String()) handleLock.Lock() defer handleLock.Unlock() @@ -81,7 +81,7 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle s } log.Log(ctx, "resolved bluesky identity", "did", ident.DID, "handle", ident.Handle, "pds", ident.PDSEndpoint()) - pdsLock := getPDSLock(ident.PDSEndpoint()) + pdsLock := pdsLocks.GetLock(ident.PDSEndpoint()) xrpcc := xrpc.Client{ Host: ident.PDSEndpoint(), Client: &aqhttp.Client, diff --git a/pkg/atproto/labeler_firehose.go b/pkg/atproto/labeler_firehose.go index 4ff14d4c..c99fc3d3 100644 --- a/pkg/atproto/labeler_firehose.go +++ b/pkg/atproto/labeler_firehose.go @@ -172,8 +172,8 @@ func (atsync *ATProtoSynchronizer) StartLabelerFirehoseRetry(ctx context.Context } targetDID = did.String() // if it's a chat message, attempt to send it to the streamers' websocket - if aturi.Collection() == "place.stream.chat.message" && l.CID != nil { - msg, err := atsync.Model.GetChatMessage(*l.CID) + if aturi.Collection() == "place.stream.chat.message" { + msg, err := atsync.Model.GetChatMessage(l.URI) if err != nil { log.Error(ctx, "failed to get chat message for label", "err", err) continue diff --git a/pkg/atproto/locks.go b/pkg/atproto/locks.go index 8304f399..2c2c22c6 100644 --- a/pkg/atproto/locks.go +++ b/pkg/atproto/locks.go @@ -1,47 +1,6 @@ package atproto -import "sync" +import "stream.place/streamplace/pkg/statedb" -// handleLocks provides per-handle synchronization -var handleLocks = struct { - sync.Mutex - locks map[string]*sync.Mutex -}{ - locks: make(map[string]*sync.Mutex), -} - -// getHandleLock returns a mutex for the given handle -func getHandleLock(handle string) *sync.Mutex { - handleLocks.Lock() - defer handleLocks.Unlock() - - if lock, exists := handleLocks.locks[handle]; exists { - return lock - } - - lock := &sync.Mutex{} - handleLocks.locks[handle] = lock - return lock -} - -// pdsLocks provides per-pds synchronization -var pdsLocks = struct { - sync.Mutex - locks map[string]*sync.Mutex -}{ - locks: make(map[string]*sync.Mutex), -} - -// getpdsLock returns a mutex for the given pds -func getPDSLock(pds string) *sync.Mutex { - pdsLocks.Lock() - defer pdsLocks.Unlock() - - if lock, exists := pdsLocks.locks[pds]; exists { - return lock - } - - lock := &sync.Mutex{} - pdsLocks.locks[pds] = lock - return lock -} +var handleLocks = statedb.NewNamedLocks() +var pdsLocks = statedb.NewNamedLocks() diff --git a/pkg/atproto/migrate.go b/pkg/atproto/migrate.go new file mode 100644 index 00000000..5c283df8 --- /dev/null +++ b/pkg/atproto/migrate.go @@ -0,0 +1,91 @@ +package atproto + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "time" + + "golang.org/x/sync/errgroup" + "stream.place/streamplace/pkg/log" +) + +func (atsync *ATProtoSynchronizer) Migrate(ctx context.Context) error { + var allDIDs []string + offset := 0 + for { + repos, err := atsync.StatefulDB.ListRepos(100, offset) + if err != nil { + return err + } + if len(repos) == 0 { + break + } + for _, repo := range repos { + allDIDs = append(allDIDs, repo.DID) + } + offset += len(repos) + } + + log.Log(ctx, "starting migration sync", "totalRepos", len(allDIDs)) + + g, ctx := errgroup.WithContext(ctx) + var syncedCount int64 + + syncErrors := map[string]error{} + syncErrorMu := sync.Mutex{} + + // Start progress logging goroutine + progressCtx, cancelProgress := context.WithCancel(ctx) + defer cancelProgress() + + go func() { + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + + for { + select { + case <-progressCtx.Done(): + return + case <-ticker.C: + current := atomic.LoadInt64(&syncedCount) + log.Log(ctx, "migration progress", "synced", current, "total", len(allDIDs)) + } + } + }() + + for i, did := range allDIDs { + currentIndex := i + currentDID := did + g.Go(func() error { + log.Debug(ctx, "syncing repo", "did", currentDID, "progress", currentIndex+1, "total", len(allDIDs)) + _, err := atsync.SyncBlueskyRepoCached(ctx, currentDID, atsync.Model) + if err != nil { + log.Error(ctx, "failed to sync repo", "did", currentDID, "err", err) + syncErrorMu.Lock() + syncErrors[currentDID] = err + syncErrorMu.Unlock() + } else { + atomic.AddInt64(&syncedCount, 1) + } + return nil + }) + } + + if err := g.Wait(); err != nil { + log.Error(ctx, "migration failed", "err", err, "synced", atomic.LoadInt64(&syncedCount), "total", len(allDIDs)) + return err + } + + for did, err := range syncErrors { + log.Error(ctx, "migration failed for user", "did", did, "err", err) + } + + if len(allDIDs) > 0 && len(syncErrors) == len(allDIDs) { + return fmt.Errorf("all users failed to migrate") + } + + log.Log(ctx, "migration completed", "synced", len(allDIDs)) + return nil +} diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index 6392feec..13dbe209 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -120,7 +120,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD if err != nil { log.Error(ctx, "failed to create chat message", "err", err) } - mcm, err = atsync.Model.GetChatMessage(cid) + mcm, err = atsync.Model.GetChatMessage(aturi.String()) if err != nil { log.Error(ctx, "failed to get just-saved chat message", "err", err) } @@ -251,7 +251,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD if rec.Reply == nil || rec.Reply.Root == nil { return nil } - livestream, err := atsync.Model.GetLivestreamByPostCID(rec.Reply.Root.Cid) + livestream, err := atsync.Model.GetLivestreamByPostURI(rec.Reply.Root.Uri) if err != nil { return fmt.Errorf("failed to get livestream: %w", err) } @@ -285,7 +285,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD RepoDID: userDID, Type: "reply", Repo: repo, - ReplyRootCID: &livestream.PostCID, + ReplyRootURI: &livestream.PostURI, ReplyRootRepoDID: &livestream.RepoDID, URI: aturi.String(), IndexedAt: &now, diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 316b9f0a..0e2dc55f 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -32,7 +32,6 @@ import ( "stream.place/streamplace/pkg/notifications" "stream.place/streamplace/pkg/replication" "stream.place/streamplace/pkg/replication/boring" - "stream.place/streamplace/pkg/resync" "stream.place/streamplace/pkg/rtmps" v0 "stream.place/streamplace/pkg/schema/v0" "stream.place/streamplace/pkg/spmetrics" @@ -216,9 +215,6 @@ func start(build *config.BuildFlags, platformJobs []jobFunc) error { } aqhttp.UserAgent = fmt.Sprintf("streamplace/%s", build.Version) - if len(os.Args) > 1 && os.Args[1] == "resync" { - return resync.Resync(ctx, &cli) - } err = os.MkdirAll(cli.DataDir, os.ModePerm) if err != nil { @@ -310,7 +306,7 @@ func start(build *config.BuildFlags, platformJobs []jobFunc) error { } var rep replication.Replicator = &boring.BoringReplicator{Peers: cli.Peers} - mod, err := model.MakeDB(cli.IndexDBPath) + mod, err := model.MakeDB(cli.DataFilePath([]string{"index"})) if err != nil { return err } @@ -321,6 +317,7 @@ func start(build *config.BuildFlags, platformJobs []jobFunc) error { return err } } + out := carstore.SQLiteStore{} err = out.Open(":memory:") if err != nil { @@ -356,6 +353,11 @@ func start(build *config.BuildFlags, platformJobs []jobFunc) error { Noter: noter, Bus: b, } + err = atsync.Migrate(ctx) + if err != nil { + return fmt.Errorf("failed to migrate: %w", err) + } + mm, err := media.MakeMediaManager(ctx, &cli, signer, rep, mod, b, atsync) if err != nil { return err diff --git a/pkg/config/config.go b/pkg/config/config.go index d6310dac..de2811ce 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -52,7 +52,6 @@ type CLI struct { Build *BuildFlags DataDir string DBURL string - IndexDBPath string EthAccountAddr string EthKeystorePath string EthPassword string @@ -128,7 +127,6 @@ func (cli *CLI) NewFlagSet(name string) *flag.FlagSet { fs.StringVar(&cli.SigningKeyPath, "signing-key", "", "Path to signing key for pushing OTA updates to the app") 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") diff --git a/pkg/model/chat_message.go b/pkg/model/chat_message.go index fd084ebf..d565ad70 100644 --- a/pkg/model/chat_message.go +++ b/pkg/model/chat_message.go @@ -14,8 +14,8 @@ import ( ) type ChatMessage struct { - CID string `json:"cid" gorm:"primaryKey;column:cid"` - URI string `json:"uri" gorm:"column:uri"` + URI string `json:"uri" gorm:"primaryKey;column:uri"` + CID string `json:"cid" gorm:"column:cid"` CreatedAt time.Time `json:"createdAt" gorm:"column:created_at;index:idx_recent_messages,priority:2"` ChatMessage *[]byte `json:"chatMessage" gorm:"column:chat_message"` RepoDID string `json:"repoDID" gorm:"column:repo_did"` @@ -83,7 +83,7 @@ func (m *DBModel) CreateChatMessage(ctx context.Context, message *ChatMessage) e return m.DB.Create(message).Error } -func (m *DBModel) GetChatMessage(cid string) (*ChatMessage, error) { +func (m *DBModel) GetChatMessage(uri string) (*ChatMessage, error) { var message ChatMessage err := m.DB. Preload("Repo"). @@ -91,7 +91,7 @@ func (m *DBModel) GetChatMessage(cid string) (*ChatMessage, error) { Preload("ReplyTo"). Preload("ReplyTo.Repo"). Preload("ReplyTo.ChatProfile"). - Where("cid = ?", cid). + Where("uri = ?", uri). First(&message). Error if errors.Is(err, gorm.ErrRecordNotFound) { diff --git a/pkg/model/feed_post.go b/pkg/model/feed_post.go index f315650b..e19bfc5b 100644 --- a/pkg/model/feed_post.go +++ b/pkg/model/feed_post.go @@ -12,15 +12,15 @@ import ( ) type FeedPost struct { - CID string `json:"cid" gorm:"primaryKey;column:cid"` - URI string `json:"uri"` + URI string `json:"uri" gorm:"primaryKey;column:uri"` + CID string `json:"cid" gorm:"column:cid"` CreatedAt time.Time `json:"createdAt" gorm:"column:created_at;index:recent_replies"` - FeedPost *[]byte `json:"feedPost"` + FeedPost *[]byte `json:"feedPost" gorm:"column:feed_post"` RepoDID string `json:"repoDID" gorm:"column:repo_did"` Repo *Repo `json:"repo,omitempty" gorm:"foreignKey:DID;references:RepoDID"` Type string `json:"type" gorm:"column:type"` - ReplyRootCID *string `json:"replyRootCID,omitempty" gorm:"column:reply_root_cid"` - ReplyRoot *FeedPost `json:"replyRoot,omitempty" gorm:"foreignKey:cid;references:ReplyRootCID"` + ReplyRootURI *string `json:"replyRootURI,omitempty" gorm:"column:reply_root_uri"` + ReplyRoot *FeedPost `json:"replyRoot,omitempty" gorm:"foreignKey:uri;references:ReplyRootURI"` ReplyRootRepoDID *string `json:"replyRootRepoDID,omitempty" gorm:"column:reply_root_repo_did;index:recent_replies"` ReplyRootRepo *Repo `json:"replyRootRepo,omitempty" gorm:"foreignKey:DID;references:ReplyRootRepoDID"` IndexedAt *time.Time `json:"indexedAt,omitempty" gorm:"column:indexed_at"` @@ -77,9 +77,9 @@ func (m *DBModel) ListFeedPostsByType(feedType string, limit int, after int64) ( return posts, nil } -func (m *DBModel) GetFeedPost(cid string) (*FeedPost, error) { +func (m *DBModel) GetFeedPost(uri string) (*FeedPost, error) { post := FeedPost{} - err := m.DB.Where("CID = ?", cid).First(&post).Error + err := m.DB.Where("uri = ?", uri).First(&post).Error if errors.Is(err, gorm.ErrRecordNotFound) { return nil, nil } diff --git a/pkg/model/livestream.go b/pkg/model/livestream.go index a7d0ab68..490d12e9 100644 --- a/pkg/model/livestream.go +++ b/pkg/model/livestream.go @@ -13,15 +13,15 @@ import ( ) type Livestream struct { - CID string `json:"cid" gorm:"primaryKey;column:cid"` - URI string `json:"uri"` + URI string `json:"uri" gorm:"primaryKey;column:uri"` + CID string `json:"cid" gorm:"column:cid"` CreatedAt time.Time `json:"createdAt" gorm:"column:created_at;index:idx_repo_created,priority:2"` Livestream *[]byte `json:"livestream"` RepoDID string `json:"repoDID" gorm:"column:repo_did;index:idx_repo_created,priority:1"` Repo *Repo `json:"repo,omitempty" gorm:"foreignKey:DID;references:RepoDID"` Post *FeedPost `json:"post,omitempty" gorm:"foreignKey:CID;references:PostCID"` - PostCID string `json:"postCID" gorm:"column:post_cid;index:idx_post_cid"` - PostURI string `json:"postURI" gorm:"column:post_uri"` + PostCID string `json:"postCID" gorm:"column:post_cid"` + PostURI string `json:"postURI" gorm:"column:post_uri;index:idx_post_uri"` } func (ls *Livestream) ToLivestreamView() (*streamplace.Livestream_LivestreamView, error) { @@ -62,18 +62,18 @@ func (m *DBModel) GetLatestLivestreamForRepo(repoDID string) (*Livestream, error return &livestream, nil } -func (m *DBModel) GetLivestreamByPostCID(postCID string) (*Livestream, error) { +func (m *DBModel) GetLivestreamByPostURI(postURI string) (*Livestream, error) { var livestream Livestream err := m.DB. Preload("Repo"). Preload("Post"). - Where("post_cid = ?", postCID). + Where("post_uri = ?", postURI). First(&livestream).Error if errors.Is(err, gorm.ErrRecordNotFound) { return nil, nil } if err != nil { - return nil, fmt.Errorf("error retrieving livestream by postCID: %w", err) + return nil, fmt.Errorf("error retrieving livestream by postURI: %w", err) } return &livestream, nil } diff --git a/pkg/model/model.go b/pkg/model/model.go index 846227c1..415d31d0 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -5,7 +5,6 @@ import ( "fmt" "os" "path/filepath" - "strings" "time" comatproto "github.com/bluesky-social/indigo/api/atproto" @@ -68,7 +67,7 @@ type Model interface { CreateLivestream(ctx context.Context, ls *Livestream) error GetLatestLivestreamForRepo(repoDID string) (*Livestream, error) - GetLivestreamByPostCID(postCID string) (*Livestream, error) + GetLivestreamByPostURI(postURI string) (*Livestream, error) GetLatestLivestreams(limit int, before *time.Time) ([]Livestream, error) CreateBlock(ctx context.Context, block *Block) error @@ -100,22 +99,24 @@ type Model interface { GetActiveLabels(uri string) ([]*comatproto.LabelDefs_Label, error) } +var DBRevision = 2 + func MakeDB(dbURL string) (Model, error) { - log.Log(context.Background(), "starting database", "dbURL", dbURL) sqliteSuffix := dbURL if dbURL != ":memory:" { - if !strings.HasPrefix(dbURL, "sqlite://") { - dbURL = fmt.Sprintf("sqlite://%s", dbURL) + // Ensure dbURL exists as a directory on the filesystem + if err := os.MkdirAll(dbURL, os.ModePerm); err != nil { + return nil, fmt.Errorf("error creating database directory: %w", err) } - sqliteSuffix := dbURL[len("sqlite://"):] + dbPath := filepath.Join(dbURL, fmt.Sprintf("index_%d.sqlite", DBRevision)) + sqliteSuffix = dbPath // if this isn't ":memory:", ensure that directory exists (eg, if db // file is being initialized) - if !strings.Contains(sqliteSuffix, ":?") { - if err := os.MkdirAll(filepath.Dir(sqliteSuffix), os.ModePerm); err != nil { - return nil, fmt.Errorf("error creating database path: %w", err) - } + if err := os.MkdirAll(filepath.Dir(sqliteSuffix), os.ModePerm); err != nil { + return nil, fmt.Errorf("error creating database path: %w", err) } } + log.Log(context.Background(), "starting database", "dbURL", sqliteSuffix) dial := sqlite.Open(sqliteSuffix) gormLogger := slogGorm.New( diff --git a/pkg/resync/resync.go b/pkg/resync/resync.go deleted file mode 100644 index 3585dc03..00000000 --- a/pkg/resync/resync.go +++ /dev/null @@ -1,113 +0,0 @@ -package resync - -import ( - "context" - "fmt" - "time" - - "golang.org/x/sync/errgroup" - "stream.place/streamplace/pkg/atproto" - "stream.place/streamplace/pkg/bus" - "stream.place/streamplace/pkg/config" - "stream.place/streamplace/pkg/log" - "stream.place/streamplace/pkg/model" -) - -// 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.IndexDBPath) - if err != nil { - return err - } - tempDBPath := cli.IndexDBPath + ".temp." + fmt.Sprintf("%d", time.Now().UnixNano()) - newMod, err := model.MakeDB(tempDBPath) - if err != nil { - return err - } - repos, err := oldMod.GetAllRepos() - if err != nil { - return err - } - - atsync := &atproto.ATProtoSynchronizer{ - 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) - - g, ctx := errgroup.WithContext(ctx) - - doneChan := make(chan string) - go func() { - for { - select { - case <-ctx.Done(): - return - case did := <-doneChan: - doneMap[did] = true - case <-time.After(10 * time.Second): - for _, repo := range repos { - if !doneMap[repo.DID] { - log.Warn(ctx, "remaining repos to sync", "did", repo.DID, "handle", repo.Handle, "pds", repo.PDS) - } - } - } - } - }() - - for _, repo := range repos { - repo := repo // capture range variable - doneMap[repo.DID] = false - g.Go(func() error { - log.Warn(ctx, "syncing repo", "did", repo.DID, "handle", repo.Handle) - ctx := log.WithLogValues(ctx, "resyncDID", repo.DID, "resyncHandle", repo.Handle) - _, err := atsync.SyncBlueskyRepoCached(ctx, repo.Handle, newMod) - if err != nil { - log.Error(ctx, "failed to sync repo", "did", repo.DID, "handle", repo.Handle, "err", err) - return nil - } - log.Log(ctx, "synced repo", "did", repo.DID, "handle", repo.Handle) - doneChan <- repo.DID - return nil - }) - } - - if err := g.Wait(); err != nil { - return err - } - - // 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)) - - // 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) - - return nil -} diff --git a/pkg/spxrpc/com_atproto_moderation.go b/pkg/spxrpc/com_atproto_moderation.go index 54d9244f..a47d1af9 100644 --- a/pkg/spxrpc/com_atproto_moderation.go +++ b/pkg/spxrpc/com_atproto_moderation.go @@ -65,7 +65,7 @@ func (s *Server) handleComAtprotoModerationCreateReport(ctx context.Context, bod did = aturi.Authority().String() // if it's chat, we want the clip from the streamer, not from the chatter if aturi.Collection() == "place.stream.chat.message" { - msg, err := s.model.GetChatMessage(body.Subject.RepoStrongRef.Cid) + msg, err := s.model.GetChatMessage(body.Subject.RepoStrongRef.Uri) if err != nil { log.Error(ctx, "failed to get chat message for chat report", "error", err) } else { diff --git a/pkg/statedb/migrate.go b/pkg/statedb/migrate.go index 34935768..f1cd94a5 100644 --- a/pkg/statedb/migrate.go +++ b/pkg/statedb/migrate.go @@ -28,7 +28,7 @@ func Migrate(cli *config.CLI) error { return err } - oldDB, err := gorm.Open(sqlite.Open(cli.IndexDBPath), &gorm.Config{ + oldDB, err := gorm.Open(sqlite.Open(cli.DataFilePath([]string{"db.sqlite"})), &gorm.Config{ Logger: gormLogger, }) if err != nil { -- 2.51.2