A fork of the Cocoon PDS but being made more distributed.
fixed cursor on startup
Signed-off-by: Will <did:plc:dadhhalkfcq3gucaq25hjqon>
authored by
willdot.net
and committed by
Tangled
(Jun 18, 2026, 5:34 PM UTC)
b46b6b05
541e1aa8
+19
-4
+17
-3
server/event_emmiter.go
+17
-3
server/event_emmiter.go
···
3
3
import (
4
4
"bytes"
5
5
"context"
6
+
"log/slog"
6
7
"net/http"
7
8
"time"
8
9
9
10
"github.com/bluesky-social/indigo/events"
10
11
"github.com/bluesky-social/indigo/lex/util"
12
+
"github.com/haileyok/cocoon/models"
11
13
)
12
14
13
15
func (s *Server) emmitEvents(ctx context.Context) error {
···
17
19
logger := s.logger.With("component", "event-emmiter")
18
20
ident := "self"
19
21
20
-
day := time.Hour * 24
22
+
// get the most recent event
23
+
rec := &models.EventRecord{}
24
+
err := s.db.Raw(ctx, "SELECT * from event_records ORDER BY seq DESC LIMIT 1", nil).Scan(&rec).Error
25
+
if err != nil {
26
+
slog.Error("fetching most recent event record", "error", err)
27
+
rec.Seq = time.Now().UnixMilli()
28
+
}
21
29
22
-
since := time.Now().Add(-day).UnixMilli()
30
+
since := rec.Seq
23
31
24
32
evts, evtManCancel, err := s.evtman.Subscribe(ctx, ident, func(evt *events.XRPCStreamEvent) bool {
25
33
return true
···
72
80
}
73
81
74
82
// TODO: use a HTTP client here not the default
75
-
_, err := http.Post(s.config.SubscribeReposServiceURL, "", buf)
83
+
resp, err := http.Post(s.config.SubscribeReposServiceURL, "", buf)
76
84
if err != nil {
77
85
logger.Error("posting to web server", "error", err)
78
86
return
79
87
}
88
+
if resp.StatusCode == http.StatusAccepted {
89
+
logger.Info("posted event to subscribe repos")
90
+
return
91
+
}
92
+
93
+
logger.Error("posting event to subscribe repos", "status", resp.StatusCode)
80
94
}()
81
95
}
82
96
+2
-1
server/persist.go
+2
-1
server/persist.go
···
116
116
}, retry.RetryIf(retryIfFunc))
117
117
118
118
if err != nil {
119
+
slog.Error(err.Error())
119
120
return err
120
121
}
121
122
···
140
141
return fmt.Errorf("failed to query events: %w", err)
141
142
}
142
143
143
-
slog.Info("playback", "len of records", len(records))
144
+
slog.Info("playback", "len of records", len(records), "cursor", cursor)
144
145
145
146
if len(records) == 0 {
146
147
return nil