ansxor.ca
forked from
tangled.org/core
this repo has no description
appview,knotclient: refactor cursor store to separate package
Signed-off-by: oppiliappan <me@oppi.li>
oppi.li
+2
-1
appview/state/knotstream.go
+2
-1
appview/state/knotstream.go
···
12
12
"tangled.sh/tangled.sh/core/appview/config"
13
13
"tangled.sh/tangled.sh/core/appview/db"
14
14
kc "tangled.sh/tangled.sh/core/knotclient"
15
+
"tangled.sh/tangled.sh/core/knotclient/cursor"
15
16
"tangled.sh/tangled.sh/core/log"
16
17
"tangled.sh/tangled.sh/core/rbac"
17
18
···
32
33
33
34
logger := log.New("knotstream")
34
35
cache := cache.New(c.Redis.Addr)
35
-
cursorStore := kc.NewRedisCursorStore(cache)
36
+
cursorStore := cursor.NewRedisCursorStore(cache)
36
37
37
38
cfg := kc.ConsumerConfig{
38
39
Sources: srcs,
+23
knotclient/cursor/memory.go
+23
knotclient/cursor/memory.go
···
1
+
package cursor
2
+
3
+
import (
4
+
"sync"
5
+
)
6
+
7
+
type MemoryStore struct {
8
+
store sync.Map
9
+
}
10
+
11
+
func (m *MemoryStore) Set(knot string, cursor int64) {
12
+
m.store.Store(knot, cursor)
13
+
}
14
+
15
+
func (m *MemoryStore) Get(knot string) (cursor int64) {
16
+
if result, ok := m.store.Load(knot); ok {
17
+
if val, ok := result.(int64); ok {
18
+
return val
19
+
}
20
+
}
21
+
22
+
return 0
23
+
}
+43
knotclient/cursor/redis.go
+43
knotclient/cursor/redis.go
···
1
+
package cursor
2
+
3
+
import (
4
+
"context"
5
+
"fmt"
6
+
"strconv"
7
+
8
+
"tangled.sh/tangled.sh/core/appview/cache"
9
+
)
10
+
11
+
const (
12
+
cursorKey = "cursor:%s"
13
+
)
14
+
15
+
type RedisStore struct {
16
+
rdb *cache.Cache
17
+
}
18
+
19
+
func NewRedisCursorStore(cache *cache.Cache) RedisStore {
20
+
return RedisStore{
21
+
rdb: cache,
22
+
}
23
+
}
24
+
25
+
func (r *RedisStore) Set(knot string, cursor int64) {
26
+
key := fmt.Sprintf(cursorKey, knot)
27
+
r.rdb.Set(context.Background(), key, cursor, 0)
28
+
}
29
+
30
+
func (r *RedisStore) Get(knot string) (cursor int64) {
31
+
key := fmt.Sprintf(cursorKey, knot)
32
+
val, err := r.rdb.Get(context.Background(), key).Result()
33
+
if err != nil {
34
+
return 0
35
+
}
36
+
cursor, err = strconv.ParseInt(val, 10, 64)
37
+
if err != nil {
38
+
// TODO: log here
39
+
return 0
40
+
}
41
+
42
+
return cursor
43
+
}
+6
knotclient/cursor/store.go
+6
knotclient/cursor/store.go
+3
-61
knotclient/events.go
+3
-61
knotclient/events.go
···
7
7
"log/slog"
8
8
"math/rand"
9
9
"net/url"
10
-
"strconv"
11
10
"sync"
12
11
"time"
13
12
14
-
"tangled.sh/tangled.sh/core/appview/cache"
13
+
"tangled.sh/tangled.sh/core/knotclient/cursor"
15
14
"tangled.sh/tangled.sh/core/log"
16
15
17
16
"github.com/gorilla/websocket"
···
36
35
QueueSize int
37
36
Logger *slog.Logger
38
37
Dev bool
39
-
CursorStore CursorStore
38
+
CursorStore cursor.Store
40
39
}
41
40
42
41
func NewConsumerConfig() *ConsumerConfig {
···
72
71
cfg ConsumerConfig
73
72
}
74
73
75
-
type CursorStore interface {
76
-
Set(knot string, cursor int64)
77
-
Get(knot string) (cursor int64)
78
-
}
79
-
80
-
type RedisCursorStore struct {
81
-
rdb *cache.Cache
82
-
}
83
-
84
-
func NewRedisCursorStore(cache *cache.Cache) RedisCursorStore {
85
-
return RedisCursorStore{
86
-
rdb: cache,
87
-
}
88
-
}
89
-
90
-
const (
91
-
cursorKey = "cursor:%s"
92
-
)
93
-
94
-
func (r *RedisCursorStore) Set(knot string, cursor int64) {
95
-
key := fmt.Sprintf(cursorKey, knot)
96
-
r.rdb.Set(context.Background(), key, cursor, 0)
97
-
}
98
-
99
-
func (r *RedisCursorStore) Get(knot string) (cursor int64) {
100
-
key := fmt.Sprintf(cursorKey, knot)
101
-
val, err := r.rdb.Get(context.Background(), key).Result()
102
-
if err != nil {
103
-
return 0
104
-
}
105
-
106
-
cursor, err = strconv.ParseInt(val, 10, 64)
107
-
if err != nil {
108
-
return 0 // optionally log parsing error
109
-
}
110
-
111
-
return cursor
112
-
}
113
-
114
-
type MemoryCursorStore struct {
115
-
store sync.Map
116
-
}
117
-
118
-
func (m *MemoryCursorStore) Set(knot string, cursor int64) {
119
-
m.store.Store(knot, cursor)
120
-
}
121
-
122
-
func (m *MemoryCursorStore) Get(knot string) (cursor int64) {
123
-
if result, ok := m.store.Load(knot); ok {
124
-
if val, ok := result.(int64); ok {
125
-
return val
126
-
}
127
-
}
128
-
129
-
return 0
130
-
}
131
-
132
74
func (e *EventConsumer) buildUrl(s EventSource, cursor int64) (*url.URL, error) {
133
75
scheme := "wss"
134
76
if e.cfg.Dev {
···
173
115
cfg.QueueSize = 100
174
116
}
175
117
if cfg.CursorStore == nil {
176
-
cfg.CursorStore = &MemoryCursorStore{}
118
+
cfg.CursorStore = &cursor.MemoryStore{}
177
119
}
178
120
return &EventConsumer{
179
121
cfg: cfg,
+1
-1
nix/vm.nix
+1
-1
nix/vm.nix
···
21
21
g = config.services.tangled-knot.gitUser;
22
22
in [
23
23
"d /var/lib/knot 0770 ${u} ${g} - -" # Create the directory first
24
-
"f+ /var/lib/knot/secret 0660 ${u} ${g} - KNOT_SERVER_SECRET=16154910ef55fe48121082c0b51fc0e360a8b15eb7bda7991d88dc9f7684427a"
24
+
"f+ /var/lib/knot/secret 0660 ${u} ${g} - KNOT_SERVER_SECRET=2650ecafdce279b09865fb1923051156eb773ee7485061b2e766086f07dbd85a"
25
25
];
26
26
services.tangled-knot = {
27
27
enable = true;