Monorepo for Tangled
appview: ingest reactions
Lewis: May this revision serve well! <lewis@tangled.org>
+76
-11
+3
-3
appview/db/reaction.go
+3
-3
appview/db/reaction.go
···
10
10
"tangled.org/core/orm"
11
11
)
12
12
13
-
func AddReaction(e Execer, reactedByDid string, threadAt syntax.ATURI, kind models.ReactionKind, rkey string) error {
14
-
query := `insert or ignore into reactions (reacted_by_did, thread_at, kind, rkey) values (?, ?, ?, ?)`
15
-
_, err := e.Exec(query, reactedByDid, threadAt, kind, rkey)
13
+
func AddReaction(e Execer, reactedByDid string, threadAt syntax.ATURI, kind models.ReactionKind, rkey string, created time.Time) error {
14
+
query := `insert or ignore into reactions (reacted_by_did, thread_at, kind, rkey, created) values (?, ?, ?, ?, ?)`
15
+
_, err := e.Exec(query, reactedByDid, threadAt, kind, rkey, created.UTC().Format(time.RFC3339))
16
16
return err
17
17
}
18
18
+57
appview/ingester.go
+57
appview/ingester.go
···
79
79
err = i.ingestVouch(ctx, e)
80
80
case tangled.FeedStarNSID:
81
81
err = i.ingestStar(ctx, e)
82
+
case tangled.FeedReactionNSID:
83
+
err = i.ingestReaction(e)
82
84
case tangled.PublicKeyNSID:
83
85
err = i.ingestPublicKey(e)
84
86
case tangled.RepoArtifactNSID:
···
1657
1659
}
1658
1660
1659
1661
return nil
1662
+
}
1663
+
1664
+
return nil
1665
+
}
1666
+
1667
+
func (i *Ingester) ingestReaction(e *jmodels.Event) error {
1668
+
did := e.Did
1669
+
rkey := e.Commit.RKey
1670
+
1671
+
l := i.Logger.With("handler", "ingestReaction", "nsid", e.Commit.Collection, "did", did, "rkey", rkey)
1672
+
l.Info("ingesting record")
1673
+
1674
+
switch e.Commit.Operation {
1675
+
case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
1676
+
raw := json.RawMessage(e.Commit.Record)
1677
+
record := tangled.FeedReaction{}
1678
+
if err := json.Unmarshal(raw, &record); err != nil {
1679
+
return fmt.Errorf("invalid record: %w", err)
1680
+
}
1681
+
1682
+
subjectUri, err := syntax.ParseATURI(record.Subject)
1683
+
if err != nil {
1684
+
return fmt.Errorf("invalid reaction subject %q: %w", record.Subject, err)
1685
+
}
1686
+
subjectUri = models.NormalizeReactionSubject(subjectUri)
1687
+
1688
+
kind, ok := models.ParseReactionKind(record.Reaction)
1689
+
if !ok {
1690
+
return fmt.Errorf("invalid reaction kind: %q", record.Reaction)
1691
+
}
1692
+
1693
+
created, parseErr := time.Parse(time.RFC3339, record.CreatedAt)
1694
+
if parseErr != nil {
1695
+
created = time.Now()
1696
+
}
1697
+
1698
+
tx, err := i.Db.Begin()
1699
+
if err != nil {
1700
+
return fmt.Errorf("failed to start transaction: %w", err)
1701
+
}
1702
+
defer tx.Rollback()
1703
+
1704
+
if err := db.DeleteReactionByRkey(tx, did, rkey); err != nil {
1705
+
return fmt.Errorf("failed to clear existing reaction: %w", err)
1706
+
}
1707
+
if err := db.AddReaction(tx, did, subjectUri, kind, rkey, created); err != nil {
1708
+
return fmt.Errorf("failed to add reaction: %w", err)
1709
+
}
1710
+
1711
+
return tx.Commit()
1712
+
1713
+
case jmodels.CommitOperationDelete:
1714
+
if err := db.DeleteReactionByRkey(i.Db, did, rkey); err != nil {
1715
+
return fmt.Errorf("failed to delete reaction record: %w", err)
1716
+
}
1660
1717
}
1661
1718
1662
1719
return nil
+11
appview/models/reaction.go
+11
appview/models/reaction.go
···
1
1
package models
2
2
3
3
import (
4
+
"fmt"
4
5
"time"
5
6
6
7
"github.com/bluesky-social/indigo/atproto/syntax"
8
+
9
+
"tangled.org/core/api/tangled"
7
10
)
8
11
9
12
type ReactionKind string
···
60
63
Count int
61
64
Users []string
62
65
}
66
+
67
+
func NormalizeReactionSubject(subject syntax.ATURI) syntax.ATURI {
68
+
switch subject.Collection() {
69
+
case tangled.RepoIssueCommentNSID, tangled.RepoPullCommentNSID:
70
+
return syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", subject.Authority(), tangled.FeedCommentNSID, subject.RecordKey()))
71
+
}
72
+
return subject
73
+
}
+4
-8
appview/state/reaction.go
+4
-8
appview/state/reaction.go
···
1
1
package state
2
2
3
3
import (
4
-
"fmt"
5
4
"net/http"
6
5
"time"
7
6
···
33
32
}
34
33
35
34
// override collection NSID to new one
36
-
switch subjectUri.Collection() {
37
-
case tangled.RepoIssueCommentNSID, tangled.RepoPullCommentNSID:
38
-
subjectUri = syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", subjectUri.Authority(), tangled.FeedCommentNSID, subjectUri.RecordKey()))
39
-
}
35
+
subjectUri = models.NormalizeReactionSubject(subjectUri)
40
36
41
37
reactionKind, ok := models.ParseReactionKind(r.URL.Query().Get("kind"))
42
38
if !ok {
···
52
48
53
49
switch r.Method {
54
50
case http.MethodPost:
55
-
createdAt := time.Now().Format(time.RFC3339)
51
+
createdAt := time.Now()
56
52
rkey := tid.TID()
57
53
resp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{
58
54
Collection: tangled.FeedReactionNSID,
···
62
58
Val: &tangled.FeedReaction{
63
59
Subject: subjectUri.String(),
64
60
Reaction: reactionKind.String(),
65
-
CreatedAt: createdAt,
61
+
CreatedAt: createdAt.Format(time.RFC3339),
66
62
},
67
63
},
68
64
})
···
71
67
return
72
68
}
73
69
74
-
err = db.AddReaction(s.db, currentUser.Did, subjectUri, reactionKind, rkey)
70
+
err = db.AddReaction(s.db, currentUser.Did, subjectUri, reactionKind, rkey, createdAt)
75
71
if err != nil {
76
72
l.Error("failed to react", "err", err)
77
73
return