tangled.org
+4
-7
knotserver/db/init.go
+4
-7
knotserver/db/init.go
···
20
create table if not exists known_dids (
21
did text primary key
22
);
23
create table if not exists public_keys (
24
id integer primary key autoincrement,
25
did text not null,
···
37
created timestamp default current_timestamp,
38
unique(did, name)
39
);
40
-
create table if not exists access_levels (
41
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
48
);
49
`)
50
if err != nil {
···
20
create table if not exists known_dids (
21
did text primary key
22
);
23
+
24
create table if not exists public_keys (
25
id integer primary key autoincrement,
26
did text not null,
···
38
created timestamp default current_timestamp,
39
unique(did, name)
40
);
41
+
42
+
create table if not exists _jetstream (
43
id integer primary key autoincrement,
44
+
last_time_us integer not null
45
);
46
`)
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
"fmt"
7
"log"
8
"net/http"
9
10
"github.com/go-chi/chi/v5"
11
tangled "github.com/sotangled/tangled/api/tangled"
···
109
collections := []string{tangled.PublicKeyNSID, tangled.KnotMemberNSID}
110
dids := []string{}
111
112
h.js = jsclient.NewJetstreamClient(collections, dids)
113
-
messages, err := h.js.ReadJetstream(ctx)
114
if err != nil {
115
return fmt.Errorf("failed to read from jetstream: %w", err)
116
}
···
150
h.e.AddMember(ThisServer, record["member"].(string))
151
}
152
default:
153
}
154
}
155
···
6
"fmt"
7
"log"
8
"net/http"
9
+
"time"
10
11
"github.com/go-chi/chi/v5"
12
tangled "github.com/sotangled/tangled/api/tangled"
···
110
collections := []string{tangled.PublicKeyNSID, tangled.KnotMemberNSID}
111
dids := []string{}
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
+
132
h.js = jsclient.NewJetstreamClient(collections, dids)
133
+
messages, err := h.js.ReadJetstream(ctx, lastTimeUs)
134
if err != nil {
135
return fmt.Errorf("failed to read from jetstream: %w", err)
136
}
···
170
h.e.AddMember(ThisServer, record["member"].(string))
171
}
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)
178
}
179
}
180
+5
-3
knotserver/jsclient/jetstream.go
+5
-3
knotserver/jsclient/jetstream.go
···
153
}
154
}
155
156
-
func (j *JetstreamClient) ReadJetstream(ctx context.Context) (chan []byte, error) {
157
-
fiveSecondsAgo := time.Now().Add(-5 * time.Second).UnixMicro()
158
159
-
if err := j.connect(fiveSecondsAgo); err != nil {
160
log.Printf("error connecting to jetstream: %v", err)
161
return nil, err
162
}
···
153
}
154
}
155
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
+
}
160
161
+
if err := j.connect(lastTimestamp); err != nil {
162
log.Printf("error connecting to jetstream: %v", err)
163
return nil, err
164
}