From 6adc18cfca4eef4867fc716a3b366ade41cb03c4 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 | 12 ++++++---- appview/ingester.go | 4 ++-- appview/models/follow.go | 9 ++++++++ appview/models/reaction.go | 9 ++++++++ appview/models/star.go | 8 +++++++ appview/state/follow.go | 46 ++++++++++++++++++++++---------------- appview/state/reaction.go | 41 +++++++++++++++++++++------------ appview/state/star.go | 45 ++++++++++++++++++++++--------------- 10 files changed, 143 insertions(+), 63 deletions(-) diff --git a/appview/db/follow.go b/appview/db/follow.go index d9d03bf7..c0e4adcd 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 f15dc711..62bccb28 100644 --- a/appview/db/reaction.go +++ b/appview/db/reaction.go @@ -9,9 +9,20 @@ import ( "tangled.org/core/appview/models" ) -func AddReaction(e Execer, did string, subjectAt syntax.ATURI, kind models.ReactionKind, rkey string) error { - query := `insert or ignore into reactions (did, subject_at, kind, rkey) values (?, ?, ?, ?)` - _, err := e.Exec(query, did, subjectAt, kind, rkey) +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 83fd00b1..03bcee4c 100644 --- a/appview/db/star.go +++ b/appview/db/star.go @@ -13,13 +13,17 @@ import ( "tangled.org/core/orm" ) -func AddStar(e Execer, star *models.Star) error { - query := `insert or ignore into stars (did, subject_at, rkey) values (?, ?, ?)` +func UpsertStar(e Execer, star models.Star) error { _, err := e.Exec( - query, + `insert into stars (did, rkey, subject_at, created) + values (?, ?, ?, ?) + on conflict(did, rkey) do update set + subject_at = excluded.subject_at, + created = excluded.created`, star.Did, - star.RepoAt.String(), star.Rkey, + star.RepoAt, + star.Created.Format(time.RFC3339), ) return err } diff --git a/appview/ingester.go b/appview/ingester.go index b92aa840..c1f7b25c 100644 --- a/appview/ingester.go +++ b/appview/ingester.go @@ -119,7 +119,7 @@ func (i *Ingester) ingestStar(e *jmodels.Event) error { l.Error("invalid record", "err", err) return err } - err = db.AddStar(i.Db, &models.Star{ + err = db.UpsertStar(i.Db, models.Star{ Did: did, RepoAt: subjectUri, Rkey: e.Commit.RKey, @@ -152,7 +152,7 @@ func (i *Ingester) ingestFollow(e *jmodels.Event) 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, diff --git a/appview/models/follow.go b/appview/models/follow.go index e9911727..d371226a 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 3cb21acf..748800f4 100644 --- a/appview/models/reaction.go +++ b/appview/models/reaction.go @@ -4,6 +4,7 @@ import ( "time" "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/api/tangled" ) type ReactionKind string @@ -56,6 +57,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/models/star.go b/appview/models/star.go index 99d450a0..b0c140d1 100644 --- a/appview/models/star.go +++ b/appview/models/star.go @@ -4,6 +4,7 @@ import ( "time" "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/api/tangled" ) type Star struct { @@ -13,6 +14,13 @@ type Star struct { Rkey string } +func (s *Star) AsRecord() tangled.FeedStar { + return tangled.FeedStar{ + Subject: s.RepoAt.String(), + CreatedAt: s.Created.Format(time.RFC3339), + } +} + // RepoStar is used for reverse mapping to repos type RepoStar struct { Star diff --git a/appview/state/follow.go b/appview/state/follow.go index 689dfe95..800393e7 100644 --- a/appview/state/follow.go +++ b/appview/state/follow.go @@ -43,38 +43,46 @@ 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.Active.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.Active.Did, - Rkey: rkey, + Rkey: follow.Rkey, Record: &lexutil.LexiconTypeDecoder{ - Val: &tangled.GraphFollow{ - Subject: subjectIdent.DID.String(), - CreatedAt: createdAt, - }}, + Val: &record, + }, }) if err != nil { log.Println("failed to create atproto record", err) return } - log.Println("created atproto record: ", resp.Uri) - follow := &models.Follow{ - UserDid: currentUser.Active.Did, - SubjectDid: subjectIdent.DID.String(), - Rkey: rkey, - } - - err = db.AddFollow(s.db, follow) - if err != nil { - log.Println("failed to follow", 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 54bc4342..0850e647 100644 --- a/appview/state/reaction.go +++ b/appview/state/reaction.go @@ -45,29 +45,44 @@ func (s *State) React(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodPost: - createdAt := time.Now().Format(time.RFC3339) - rkey := tid.TID() + reaction := models.Reaction{ + ReactedByDid: currentUser.Active.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 { + log.Println("failed to react", err) + return + } + + record := reaction.AsRecord() resp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ Collection: tangled.FeedReactionNSID, Repo: currentUser.Active.Did, - Rkey: rkey, + Rkey: reaction.Rkey, Record: &lexutil.LexiconTypeDecoder{ - Val: &tangled.FeedReaction{ - Subject: subjectUri.String(), - Reaction: reactionKind.String(), - CreatedAt: createdAt, - }, + Val: &record, }, }) if err != nil { log.Println("failed to create atproto record", err) return } + log.Println("created atproto record: ", resp.Uri) - err = db.AddReaction(s.db, currentUser.Active.Did, subjectUri, reactionKind, rkey) - if err != nil { - log.Println("failed to react", 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) @@ -75,8 +90,6 @@ func (s *State) React(w http.ResponseWriter, r *http.Request) { log.Println("failed to get reactions for ", subjectUri) } - log.Println("created atproto record: ", resp.Uri) - s.pages.ThreadReactionFragment(w, pages.ThreadReactionFragmentParams{ ThreadAt: subjectUri, Kind: reactionKind, diff --git a/appview/state/star.go b/appview/state/star.go index e150e9de..101ba6f9 100644 --- a/appview/state/star.go +++ b/appview/state/star.go @@ -38,17 +38,33 @@ 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.Active.Did, + Rkey: tid.TID(), + RepoAt: 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.UpsertStar(tx, star); err != nil { + s.logger.Error("failed to star", "err", err) + return + } + + record := star.AsRecord() resp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ Collection: tangled.FeedStarNSID, Repo: currentUser.Active.Did, - Rkey: rkey, + Rkey: star.Rkey, Record: &lexutil.LexiconTypeDecoder{ - Val: &tangled.FeedStar{ - Subject: subjectUri.String(), - CreatedAt: createdAt, - }}, + Val: &record, + }, }) if err != nil { log.Println("failed to create atproto record", err) @@ -56,25 +72,18 @@ func (s *State) Star(w http.ResponseWriter, r *http.Request) { } log.Println("created atproto record: ", resp.Uri) - star := &models.Star{ - Did: currentUser.Active.Did, - RepoAt: subjectUri, - Rkey: rkey, + 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 } - err = db.AddStar(s.db, star) - if err != nil { - log.Println("failed to star", err) - return - } + s.notifier.NewStar(r.Context(), &star) starCount, err := db.GetStarCount(s.db, subjectUri) if err != nil { log.Println("failed to get star count for ", subjectUri) } - s.notifier.NewStar(r.Context(), star) - s.pages.StarBtnFragment(w, pages.StarBtnFragmentParams{ IsStarred: true, SubjectAt: subjectUri, -- 2.51.2