diff --git a/appview/db/follow.go b/appview/db/follow.go index f41de085..86012dee 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 62af9b13..bb223e48 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 e4d7bb65..7ee5cf3c 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 5a0ef070..9b37c60d 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 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 d24845bd..171b6e14 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 8de8f972..341ce5f0 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 d58f4a82..d2196435 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 03ff587b..ffcade19 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 })