From ed6a383bb9dc70d4d7eb5ffd8770957c9ef1c1d4 Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Fri, 16 Jan 2026 19:41:29 +0900 Subject: [PATCH] appview: upsert star/reaction/follow records Most service flow will be: 1. start db transaction 2. run db operation 3. run PDS operation 4. rollback db if anything above failed 5. commit transaction If PDS operation succeed, don't try rollback anymore. The ingester will backfill the missed db operations. Signed-off-by: Seongmin Lee --- appview/db/follow.go | 15 ++++++++--- appview/db/reaction.go | 17 ++++++++++--- appview/db/star.go | 13 +++++++--- appview/ingester.go | 30 +++++++++------------- appview/models/follow.go | 9 +++++++ appview/models/reaction.go | 8 ++++++ appview/state/follow.go | 45 +++++++++++++++++++-------------- appview/state/reaction.go | 43 +++++++++++++++++++++----------- appview/state/star.go | 51 ++++++++++++++++++++++---------------- 9 files changed, 149 insertions(+), 82 deletions(-) diff --git a/appview/db/follow.go b/appview/db/follow.go index f41de0855..86012dee4 100644 --- a/appview/db/follow.go +++ b/appview/db/follow.go @@ -11,9 +11,18 @@ import ( "tangled.org/core/orm" ) -func AddFollow(e Execer, follow *models.Follow) error { - query := `insert or ignore into follows (did, subject_did, rkey) values (?, ?, ?)` - _, err := e.Exec(query, follow.UserDid, follow.SubjectDid, follow.Rkey) +func UpsertFollow(e Execer, follow models.Follow) error { + _, err := e.Exec( + `insert into follows (did, rkey, subject_did, created) + values (?, ?, ?, ?) + on conflict(did, rkey) do update set + subject_did = excluded.subject_did, + created = excluded.created`, + follow.UserDid, + follow.Rkey, + follow.SubjectDid, + follow.FollowedAt.Format(time.RFC3339), + ) return err } diff --git a/appview/db/reaction.go b/appview/db/reaction.go index 62af9b135..bb223e48f 100644 --- a/appview/db/reaction.go +++ b/appview/db/reaction.go @@ -10,9 +10,20 @@ import ( "tangled.org/core/orm" ) -func AddReaction(e Execer, did string, subjectAt syntax.ATURI, kind models.ReactionKind, rkey string, created time.Time) error { - query := `insert or ignore into reactions (did, subject_at, kind, rkey, created) values (?, ?, ?, ?, ?)` - _, err := e.Exec(query, did, subjectAt, kind, rkey, created.UTC().Format(time.RFC3339)) +func UpsertReaction(e Execer, reaction models.Reaction) error { + _, err := e.Exec( + `insert into reactions (did, rkey, subject_at, kind, created) + values (?, ?, ?, ?, ?) + on conflict(did, rkey) do update set + subject_at = excluded.subject_at, + kind = excluded.kind, + created = excluded.created`, + reaction.ReactedByDid, + reaction.Rkey, + reaction.ThreadAt, + reaction.Kind, + reaction.Created.Format(time.RFC3339), + ) return err } diff --git a/appview/db/star.go b/appview/db/star.go index e4d7bb651..7ee5cf3c7 100644 --- a/appview/db/star.go +++ b/appview/db/star.go @@ -13,14 +13,19 @@ import ( "tangled.org/core/orm" ) -func AddStar(e Execer, star *models.Star) error { - query := `insert or ignore into stars (did, subject_type, subject, rkey) values (?, ?, ?, ?)` +func UpsertStar(e Execer, star models.Star) error { _, err := e.Exec( - query, + `insert into stars (did, rkey, subject_type, subject, created) + values (?, ?, ?, ?, ?) + on conflict(did, rkey) do update set + subject_type = excluded.subject_type, + subject = excluded.subject, + created = excluded.created`, star.Did, + star.Rkey, string(star.SubjectType), star.Subject, - star.Rkey, + star.Created.Format(time.RFC3339), ) return err } diff --git a/appview/ingester.go b/appview/ingester.go index 5a0ef0703..9b37c60d2 100644 --- a/appview/ingester.go +++ b/appview/ingester.go @@ -200,14 +200,14 @@ func (i *Ingester) ingestStar(ctx context.Context, e *jmodels.Event, l *slog.Log record := tangled.FeedStar{} unmarshalErr := json.Unmarshal(raw, &record) - star := &models.Star{ + star := models.Star{ Did: did, Rkey: e.Commit.RKey, } switch { case unmarshalErr != nil: - resolved, resolveErr := i.resolveOldFormatStar(raw, star, l) + resolved, resolveErr := i.resolveOldFormatStar(raw, &star, l) if resolveErr != nil { l.Error("invalid record", "newFmtErr", unmarshalErr, "oldFmtErr", resolveErr) return unmarshalErr @@ -236,7 +236,7 @@ func (i *Ingester) ingestStar(ctx context.Context, e *jmodels.Event, l *slog.Log return fmt.Errorf("star record has empty subject union") } - err = db.AddStar(i.Db, star) + err = db.UpsertStar(i.Db, star) case jmodels.CommitOperationDelete: err = db.DeleteStarByRkey(i.Db, did, e.Commit.RKey) } @@ -265,7 +265,7 @@ func (i *Ingester) ingestFollow(e *jmodels.Event, l *slog.Logger) error { return err } - err = db.AddFollow(i.Db, &models.Follow{ + err = db.UpsertFollow(i.Db, models.Follow{ UserDid: did, SubjectDid: record.Subject, Rkey: e.Commit.RKey, @@ -1733,21 +1733,15 @@ func (i *Ingester) ingestReaction(e *jmodels.Event, l *slog.Logger) error { created = time.Now() } - tx, err := i.Db.Begin() - if err != nil { - return fmt.Errorf("failed to start transaction: %w", err) - } - defer tx.Rollback() - - if err := db.DeleteReactionByRkey(tx, did, rkey); err != nil { - return fmt.Errorf("failed to clear existing reaction: %w", err) - } - if err := db.AddReaction(tx, did, subjectUri, kind, rkey, created); err != nil { - return fmt.Errorf("failed to add reaction: %w", err) + reaction := models.Reaction{ + ReactedByDid: did, + Rkey: rkey, + ThreadAt: subjectUri, + Kind: kind, + Created: created, } - - if err := tx.Commit(); err != nil { - return err + if err := db.UpsertReaction(i.Db, reaction); err != nil { + return fmt.Errorf("failed to upsert reaction: %w", err) } case jmodels.CommitOperationDelete: diff --git a/appview/models/follow.go b/appview/models/follow.go index e99117271..d371226ac 100644 --- a/appview/models/follow.go +++ b/appview/models/follow.go @@ -2,6 +2,8 @@ package models import ( "time" + + "tangled.org/core/api/tangled" ) type Follow struct { @@ -11,6 +13,13 @@ type Follow struct { Rkey string } +func (f *Follow) AsRecord() tangled.GraphFollow { + return tangled.GraphFollow{ + Subject: f.SubjectDid, + CreatedAt: f.FollowedAt.Format(time.RFC3339), + } +} + type FollowStats struct { Followers int64 Following int64 diff --git a/appview/models/reaction.go b/appview/models/reaction.go index d24845bdb..171b6e142 100644 --- a/appview/models/reaction.go +++ b/appview/models/reaction.go @@ -59,6 +59,14 @@ type Reaction struct { Kind ReactionKind } +func (r *Reaction) AsRecord() tangled.FeedReaction { + return tangled.FeedReaction{ + Subject: r.ThreadAt.String(), + Reaction: r.Kind.String(), + CreatedAt: r.Created.Format(time.RFC3339), + } +} + type ReactionDisplayData struct { Count int Users []string diff --git a/appview/state/follow.go b/appview/state/follow.go index 8de8f9726..341ce5f0d 100644 --- a/appview/state/follow.go +++ b/appview/state/follow.go @@ -43,17 +43,33 @@ func (s *State) Follow(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodPost: - createdAt := time.Now().Format(time.RFC3339) - rkey := tid.TID() + follow := models.Follow{ + UserDid: currentUser.Did, + SubjectDid: subjectIdent.DID.String(), + Rkey: tid.TID(), + FollowedAt: time.Now(), + } + + tx, err := s.db.BeginTx(r.Context(), nil) + if err != nil { + s.logger.Error("failed to start transaction", "err", err) + return + } + defer tx.Rollback() + + if err := db.UpsertFollow(tx, follow); err != nil { + s.logger.Error("failed to follow", "err", err) + return + } + + record := follow.AsRecord() resp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ Collection: tangled.GraphFollowNSID, Repo: currentUser.Did, - Rkey: rkey, + Rkey: follow.Rkey, Record: &lexutil.LexiconTypeDecoder{ - Val: &tangled.GraphFollow{ - Subject: subjectIdent.DID.String(), - CreatedAt: createdAt, - }}, + Val: &record, + }, }) if err != nil { l.Error("failed to create atproto record", "err", err) @@ -62,19 +78,12 @@ func (s *State) Follow(w http.ResponseWriter, r *http.Request) { l.Info("created atproto record", "uri", resp.Uri) - follow := &models.Follow{ - UserDid: currentUser.Did, - SubjectDid: subjectIdent.DID.String(), - Rkey: rkey, - } - - err = db.AddFollow(s.db, follow) - if err != nil { - l.Error("failed to follow", "err", err) - return + if err := tx.Commit(); err != nil { + s.logger.Error("failed to commit transaction", "err", err) + // DB op failed but record is created in PDS. Ingester will backfill the missed operation } - s.notifier.NewFollow(r.Context(), follow) + s.notifier.NewFollow(r.Context(), &follow) followStats, err := db.GetFollowerFollowingCount(s.db, subjectIdent.DID.String()) if err != nil { diff --git a/appview/state/reaction.go b/appview/state/reaction.go index d58f4a823..d21964356 100644 --- a/appview/state/reaction.go +++ b/appview/state/reaction.go @@ -48,38 +48,51 @@ func (s *State) React(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodPost: - createdAt := time.Now() - rkey := tid.TID() + reaction := models.Reaction{ + ReactedByDid: currentUser.Did, + Rkey: tid.TID(), + Kind: reactionKind, + ThreadAt: subjectUri, + Created: time.Now(), + } + + tx, err := s.db.BeginTx(r.Context(), nil) + if err != nil { + s.logger.Error("failed to start transaction", "err", err) + return + } + defer tx.Rollback() + + if err := db.UpsertReaction(tx, reaction); err != nil { + l.Error("db: failed to upsert reaction", "err", err) + return + } + + record := reaction.AsRecord() resp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ Collection: tangled.FeedReactionNSID, Repo: currentUser.Did, - Rkey: rkey, + Rkey: reaction.Rkey, Record: &lexutil.LexiconTypeDecoder{ - Val: &tangled.FeedReaction{ - Subject: subjectUri.String(), - Reaction: reactionKind.String(), - CreatedAt: createdAt.Format(time.RFC3339), - }, + Val: &record, }, }) if err != nil { l.Error("failed to create atproto record", "err", err) return } + l.Info("created atproto record", "uri", resp.Uri) - err = db.AddReaction(s.db, currentUser.Did, subjectUri, reactionKind, rkey, createdAt) - if err != nil { - l.Error("failed to react", "err", err) - return + if err := tx.Commit(); err != nil { + s.logger.Error("failed to commit transaction", "err", err) + // DB op failed but record is created in PDS. Ingester will backfill the missed operation } reactionMap, err := db.GetReactionMap(s.db, 20, subjectUri) if err != nil { - l.Error("failed to get reactions", "subjectUri", subjectUri, "err", err) + l.Error("failed to get reactions", "subject", subjectUri) } - l.Info("created atproto record", "uri", resp.Uri) - s.pages.ThreadReactionFragment(w, pages.ThreadReactionFragmentParams{ Kind: reactionKind, Count: reactionMap[reactionKind].Count, diff --git a/appview/state/star.go b/appview/state/star.go index 03ff587b8..ffcade199 100644 --- a/appview/state/star.go +++ b/appview/state/star.go @@ -77,19 +77,36 @@ func (s *State) Star(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodPost: - createdAt := time.Now().Format(time.RFC3339) - rkey := tid.TID() + star := models.Star{ + Did: currentUser.Did, + Rkey: tid.TID(), + SubjectType: subjectType, + Subject: subjectKey, + Created: time.Now(), + } + + tx, err := s.db.BeginTx(r.Context(), nil) + if err != nil { + l.Error("failed to start transaction", "err", err) + return + } + defer tx.Rollback() - starRecord := &tangled.FeedStar{ - CreatedAt: createdAt, - Subject: starSubject, + if err := db.UpsertStar(tx, star); err != nil { + l.Error("failed to star", "err", err) + return } resp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ Collection: tangled.FeedStarNSID, Repo: currentUser.Did, - Rkey: rkey, - Record: &lexutil.LexiconTypeDecoder{Val: starRecord}, + Rkey: star.Rkey, + Record: &lexutil.LexiconTypeDecoder{ + Val: &tangled.FeedStar{ + CreatedAt: star.Created.Format(time.RFC3339), + Subject: starSubject, + }, + }, }) if err != nil { l.Error("failed to create atproto record", "err", err) @@ -97,26 +114,18 @@ func (s *State) Star(w http.ResponseWriter, r *http.Request) { } l.Info("created atproto record", "uri", resp.Uri) - star := &models.Star{ - Did: currentUser.Did, - SubjectType: subjectType, - Subject: subjectKey, - Rkey: rkey, + if err := tx.Commit(); err != nil { + l.Error("failed to commit transaction", "err", err) + // DB op failed but record is created in PDS. Ingester will backfill the missed operation } - err = db.AddStar(s.db, star) - if err != nil { - l.Error("failed to star", "err", err) - return - } + s.notifier.NewStar(r.Context(), &star) starCount, err := db.GetStarCount(s.db, subjectType, subjectKey) if err != nil { l.Error("failed to get star count", "subject", subjectKey, "err", err) } - s.notifier.NewStar(r.Context(), star) - s.pages.StarBtnFragment(w, pages.StarBtnFragmentParams{ IsStarred: true, SubjectAt: subjectUri, @@ -162,9 +171,9 @@ func (s *State) Star(w http.ResponseWriter, r *http.Request) { } s.notifier.DeleteStar(r.Context(), &models.Star{ - Did: currentUser.Did, + Did: currentUser.Did, SubjectType: subjectType, - Subject: subjectKey, + Subject: subjectKey, // Rkey // Created }) -- 2.51.2