bad-example.com
forked from
tangled.org/core
this repo has no description
knotserver/jetstream: start jetstream watcher from last time_us
If last time_us is older than a week, we discard it and start over.
This also refreshes the saved last time_us in the database.
anirudh.fi
+48
-11
+4
-7
knotserver/db/init.go
+4
-7
knotserver/db/init.go
···
20
20
create table if not exists known_dids (
21
21
did text primary key
22
22
);
23
+
23
24
create table if not exists public_keys (
24
25
id integer primary key autoincrement,
25
26
did text not null,
···
37
38
created timestamp default current_timestamp,
38
39
unique(did, name)
39
40
);
40
-
create table if not exists access_levels (
41
+
42
+
create table if not exists _jetstream (
41
43
id integer primary key autoincrement,
42
-
repo_id integer not null,
43
-
did text not null,
44
-
access text not null check (access in ('OWNER', 'WRITER')),
45
-
created timestamp default current_timestamp,
46
-
unique(repo_id, did),
47
-
foreign key (repo_id) references repos(id) on delete cascade
44
+
last_time_us integer not null
48
45
);
49
46
`)
50
47
if err != nil {
+13
knotserver/db/jetstream.go
+13
knotserver/db/jetstream.go
···
1
+
package db
2
+
3
+
func (d *DB) SaveLastTimeUs(lastTimeUs int64) error {
4
+
_, err := d.db.Exec(`insert into _jetstream (last_time_us) values (?)`, lastTimeUs)
5
+
return err
6
+
}
7
+
8
+
func (d *DB) GetLastTimeUs() (int64, error) {
9
+
var lastTimeUs int64
10
+
row := d.db.QueryRow(`select last_time_us from _jetstream`)
11
+
err := row.Scan(&lastTimeUs)
12
+
return lastTimeUs, err
13
+
}
+26
-1
knotserver/handler.go
+26
-1
knotserver/handler.go
···
6
6
"fmt"
7
7
"log"
8
8
"net/http"
9
+
"time"
9
10
10
11
"github.com/go-chi/chi/v5"
11
12
tangled "github.com/sotangled/tangled/api/tangled"
···
109
110
collections := []string{tangled.PublicKeyNSID, tangled.KnotMemberNSID}
110
111
dids := []string{}
111
112
113
+
var lastTimeUs int64
114
+
var err error
115
+
lastTimeUs, err = h.db.GetLastTimeUs()
116
+
if err != nil {
117
+
log.Println("couldn't get last time us, starting from now")
118
+
lastTimeUs = time.Now().UnixMicro()
119
+
}
120
+
// If last time is older than a week, start from now
121
+
if time.Now().UnixMicro()-lastTimeUs > 7*24*60*60*1000*1000 {
122
+
lastTimeUs = time.Now().UnixMicro()
123
+
log.Printf("last time us is older than a week. discarding that and starting from now.")
124
+
err = h.db.SaveLastTimeUs(lastTimeUs)
125
+
if err != nil {
126
+
log.Println("failed to save last time us")
127
+
}
128
+
}
129
+
130
+
log.Printf("found last time_us %d", lastTimeUs)
131
+
112
132
h.js = jsclient.NewJetstreamClient(collections, dids)
113
-
messages, err := h.js.ReadJetstream(ctx)
133
+
messages, err := h.js.ReadJetstream(ctx, lastTimeUs)
114
134
if err != nil {
115
135
return fmt.Errorf("failed to read from jetstream: %w", err)
116
136
}
···
150
170
h.e.AddMember(ThisServer, record["member"].(string))
151
171
}
152
172
default:
173
+
}
174
+
175
+
lastTimeUs := int64(data["time_us"].(float64))
176
+
if err := h.db.SaveLastTimeUs(lastTimeUs); err != nil {
177
+
log.Printf("failed to save last time us: %v", err)
153
178
}
154
179
}
155
180
+5
-3
knotserver/jsclient/jetstream.go
+5
-3
knotserver/jsclient/jetstream.go
···
153
153
}
154
154
}
155
155
156
-
func (j *JetstreamClient) ReadJetstream(ctx context.Context) (chan []byte, error) {
157
-
fiveSecondsAgo := time.Now().Add(-5 * time.Second).UnixMicro()
156
+
func (j *JetstreamClient) ReadJetstream(ctx context.Context, lastTimestamp int64) (chan []byte, error) {
157
+
if lastTimestamp == 0 {
158
+
lastTimestamp = time.Now().Add(-5 * time.Second).UnixMicro()
159
+
}
158
160
159
-
if err := j.connect(fiveSecondsAgo); err != nil {
161
+
if err := j.connect(lastTimestamp); err != nil {
160
162
log.Printf("error connecting to jetstream: %v", err)
161
163
return nil, err
162
164
}