A fork of the Cocoon PDS but being made more distributed.
event sequencing
Signed-off-by: Will <did:plc:dadhhalkfcq3gucaq25hjqon>
willdot.net
(Jun 6, 2026, 9:19 AM +0100)
b09fc7a1
20ed5d21
+16
-28
+16
-28
server/persist.go
+16
-28
server/persist.go
···
4
4
"bytes"
5
5
"context"
6
6
"fmt"
7
+
"log/slog"
7
8
"sync"
8
9
"time"
9
10
···
19
20
type DbPersister struct {
20
21
Db *gorm.DB
21
22
22
-
Lk sync.Mutex
23
-
Seq int64
23
+
Lk sync.Mutex
24
24
25
25
Broadcast func(*events.XRPCStreamEvent)
26
26
···
42
42
Retention: retention,
43
43
}
44
44
45
-
// kind of hacky. we will try and get the latest one from the db, but if it doesn't exist...well we have a problem
46
-
// because the relay will already have _some_ value > 0 set as a cursor, we'll want to just set this to some high value
47
-
// we'll just grab a current unix timestamp and set that as the cursor
48
-
var lastEvent models.EventRecord
49
-
if err := db.Order("seq desc").Limit(1).First(&lastEvent).Error; err != nil {
50
-
if err != gorm.ErrRecordNotFound {
51
-
return nil, fmt.Errorf("failed to get last event seq: %w", err)
52
-
}
53
-
p.Seq = time.Now().Unix()
54
-
} else {
55
-
p.Seq = lastEvent.Seq
56
-
}
57
-
58
45
go p.cleanupRoutine()
59
46
60
47
return p, nil
···
68
55
p.Lk.Lock()
69
56
defer p.Lk.Unlock()
70
57
71
-
p.Seq++
72
-
seq := p.Seq
58
+
rec := &models.EventRecord{}
59
+
if err := p.Db.Order("seq desc").Limit(1).First(rec).Error; err != nil {
60
+
slog.Error("fetching most recent event record", "error", err)
61
+
rec.Seq = time.Now().Unix()
62
+
}
63
+
rec.Seq++
73
64
74
65
var did string
75
66
var evtType string
76
67
77
68
switch {
78
69
case e.RepoCommit != nil:
79
-
e.RepoCommit.Seq = seq
70
+
e.RepoCommit.Seq = rec.Seq
80
71
did = e.RepoCommit.Repo
81
72
evtType = "commit"
82
73
case e.RepoSync != nil:
83
-
e.RepoSync.Seq = seq
74
+
e.RepoSync.Seq = rec.Seq
84
75
did = e.RepoSync.Did
85
76
evtType = "sync"
86
77
case e.RepoIdentity != nil:
87
-
e.RepoIdentity.Seq = seq
78
+
e.RepoIdentity.Seq = rec.Seq
88
79
did = e.RepoIdentity.Did
89
80
evtType = "identity"
90
81
case e.RepoAccount != nil:
91
-
e.RepoAccount.Seq = seq
82
+
e.RepoAccount.Seq = rec.Seq
92
83
did = e.RepoAccount.Did
93
84
evtType = "account"
94
85
default:
···
100
91
return fmt.Errorf("failed to serialize event: %w", err)
101
92
}
102
93
103
-
rec := &models.EventRecord{
104
-
Seq: seq,
105
-
CreatedAt: time.Now(),
106
-
Did: did,
107
-
Type: evtType,
108
-
Data: data,
109
-
}
94
+
rec.CreatedAt = time.Now()
95
+
rec.Did = did
96
+
rec.Type = evtType
97
+
rec.Data = data
110
98
111
99
if err := p.Db.Create(rec).Error; err != nil {
112
100
return fmt.Errorf("failed to persist event: %w", err)