From 85720a2cf58eff0c17feef0b2b887f790462d4cf Mon Sep 17 00:00:00 2001 From: oppiliappan Date: Tue, 17 Jun 2025 10:21:07 +0000 Subject: [PATCH] eventconsumer: extract knotclient/events into its own package Signed-off-by: oppiliappan --- eventconsumer/consumer.go | 245 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ eventconsumer/knot.go | 39 +++++++++++++++++++++++++++++++++++++++ eventconsumer/spindle.go | 39 +++++++++++++++++++++++++++++++++++++++ knotclient/events.go | 268 ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- spindle/ingester.go | 4 ++-- spindle/server.go | 16 ++++++++-------- spindle/stream.go | 29 ++++++++++++++++++++++++----- eventconsumer/cursor/memory.go | 23 +++++++++++++++++++++++ eventconsumer/cursor/redis.go | 43 +++++++++++++++++++++++++++++++++++++++++++ eventconsumer/cursor/sqlite.go | 83 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ eventconsumer/cursor/store.go | 6 ++++++ knotclient/cursor/memory.go | 23 ----------------------- knotclient/cursor/redis.go | 43 ------------------------------------------- knotclient/cursor/sqlite.go | 83 ----------------------------------------------------------------------------------- knotclient/cursor/store.go | 6 ------ 15 file(s) changed, 512 insertion(s)(+), 438 deletion(s)(-) diff --git a/eventconsumer/consumer.go b/eventconsumer/consumer.go new file mode 100644 --- /dev/null +++ b/eventconsumer/consumer.go @@ -0,0 +1,245 @@ +package eventconsumer + +import ( + "context" + "encoding/json" + "log/slog" + "math/rand" + "net/url" + "sync" + "time" + + "tangled.sh/tangled.sh/core/eventconsumer/cursor" + "tangled.sh/tangled.sh/core/log" + + "github.com/gorilla/websocket" +) + +type ProcessFunc func(ctx context.Context, source Source, message Message) error + +type Message struct { + Rkey string + Nsid string + // do not full deserialize this portion of the message, processFunc can do that + EventJson json.RawMessage `json:"event"` +} + +type ConsumerConfig struct { + Sources map[Source]struct{} + ProcessFunc ProcessFunc + RetryInterval time.Duration + MaxRetryInterval time.Duration + ConnectionTimeout time.Duration + WorkerCount int + QueueSize int + Logger *slog.Logger + Dev bool + CursorStore cursor.Store +} + +func NewConsumerConfig() *ConsumerConfig { + return &ConsumerConfig{ + Sources: make(map[Source]struct{}), + } +} + +type Source interface { + // url to start streaming events from + Url(cursor int64, dev bool) (*url.URL, error) + // cache key for cursor storage + Key() string +} + +type Consumer struct { + wg sync.WaitGroup + dialer *websocket.Dialer + connMap sync.Map + jobQueue chan job + logger *slog.Logger + randSource *rand.Rand + + // rw lock over edits to ConsumerConfig + cfgMu sync.RWMutex + cfg ConsumerConfig +} + +type job struct { + source Source + message []byte +} + +func NewConsumer(cfg ConsumerConfig) *Consumer { + if cfg.RetryInterval == 0 { + cfg.RetryInterval = 15 * time.Minute + } + if cfg.ConnectionTimeout == 0 { + cfg.ConnectionTimeout = 10 * time.Second + } + if cfg.WorkerCount <= 0 { + cfg.WorkerCount = 5 + } + if cfg.MaxRetryInterval == 0 { + cfg.MaxRetryInterval = 1 * time.Hour + } + if cfg.Logger == nil { + cfg.Logger = log.New("consumer") + } + if cfg.QueueSize == 0 { + cfg.QueueSize = 100 + } + if cfg.CursorStore == nil { + cfg.CursorStore = &cursor.MemoryStore{} + } + return &Consumer{ + cfg: cfg, + dialer: websocket.DefaultDialer, + jobQueue: make(chan job, cfg.QueueSize), // buffered job queue + logger: cfg.Logger, + randSource: rand.New(rand.NewSource(time.Now().UnixNano())), + } +} + +func (c *Consumer) Start(ctx context.Context) { + c.cfg.Logger.Info("starting consumer", "config", c.cfg) + + // start workers + for range c.cfg.WorkerCount { + c.wg.Add(1) + go c.worker(ctx) + } + + // start streaming + for source := range c.cfg.Sources { + c.wg.Add(1) + go c.startConnectionLoop(ctx, source) + } +} + +func (c *Consumer) Stop() { + c.connMap.Range(func(_, val any) bool { + if conn, ok := val.(*websocket.Conn); ok { + conn.Close() + } + return true + }) + c.wg.Wait() + close(c.jobQueue) +} + +func (c *Consumer) AddSource(ctx context.Context, s Source) { + // we are already listening to this source + if _, ok := c.cfg.Sources[s]; ok { + c.logger.Info("source already present", "source", s) + return + } + + c.cfgMu.Lock() + c.cfg.Sources[s] = struct{}{} + c.wg.Add(1) + go c.startConnectionLoop(ctx, s) + c.cfgMu.Unlock() +} + +func (c *Consumer) worker(ctx context.Context) { + defer c.wg.Done() + for { + select { + case <-ctx.Done(): + return + case j, ok := <-c.jobQueue: + if !ok { + return + } + + var msg Message + err := json.Unmarshal(j.message, &msg) + if err != nil { + c.logger.Error("error deserializing message", "source", j.source.Key(), "err", err) + return + } + + // update cursor + c.cfg.CursorStore.Set(j.source.Key(), time.Now().UnixNano()) + + if err := c.cfg.ProcessFunc(ctx, j.source, msg); err != nil { + c.logger.Error("error processing message", "source", j.source, "err", err) + } + } + } +} + +func (c *Consumer) startConnectionLoop(ctx context.Context, source Source) { + defer c.wg.Done() + retryInterval := c.cfg.RetryInterval + for { + select { + case <-ctx.Done(): + return + default: + err := c.runConnection(ctx, source) + if err != nil { + c.logger.Error("connection failed", "source", source, "err", err) + } + + // apply jitter + jitter := time.Duration(c.randSource.Int63n(int64(retryInterval) / 5)) + delay := retryInterval + jitter + + if retryInterval < c.cfg.MaxRetryInterval { + retryInterval *= 2 + if retryInterval > c.cfg.MaxRetryInterval { + retryInterval = c.cfg.MaxRetryInterval + } + } + c.logger.Info("retrying connection", "source", source, "delay", delay) + select { + case <-time.After(delay): + case <-ctx.Done(): + return + } + } + } +} + +func (c *Consumer) runConnection(ctx context.Context, source Source) error { + connCtx, cancel := context.WithTimeout(ctx, c.cfg.ConnectionTimeout) + defer cancel() + + cursor := c.cfg.CursorStore.Get(source.Key()) + + u, err := source.Url(cursor, c.cfg.Dev) + if err != nil { + return err + } + + c.logger.Info("connecting", "url", u.String()) + conn, _, err := c.dialer.DialContext(connCtx, u.String(), nil) + if err != nil { + return err + } + defer conn.Close() + c.connMap.Store(source, conn) + defer c.connMap.Delete(source) + + c.logger.Info("connected", "source", source) + + for { + select { + case <-ctx.Done(): + return nil + default: + msgType, msg, err := conn.ReadMessage() + if err != nil { + return err + } + if msgType != websocket.TextMessage { + continue + } + select { + case c.jobQueue <- job{source: source, message: msg}: + case <-ctx.Done(): + return nil + } + } + } +} diff --git a/eventconsumer/knot.go b/eventconsumer/knot.go new file mode 100644 --- /dev/null +++ b/eventconsumer/knot.go @@ -0,0 +1,39 @@ +package eventconsumer + +import ( + "fmt" + "net/url" +) + +type KnotSource struct { + Knot string +} + +func (k KnotSource) Key() string { + return k.Knot +} + +func (k KnotSource) Url(cursor int64, dev bool) (*url.URL, error) { + scheme := "wss" + if dev { + scheme = "ws" + } + + u, err := url.Parse(scheme + "://" + k.Knot + "/events") + if err != nil { + return nil, err + } + + if cursor != 0 { + query := url.Values{} + query.Add("cursor", fmt.Sprintf("%d", cursor)) + u.RawQuery = query.Encode() + } + return u, nil +} + +func NewKnotSource(knot string) KnotSource { + return KnotSource{ + Knot: knot, + } +} diff --git a/eventconsumer/spindle.go b/eventconsumer/spindle.go new file mode 100644 --- /dev/null +++ b/eventconsumer/spindle.go @@ -0,0 +1,39 @@ +package eventconsumer + +import ( + "fmt" + "net/url" +) + +type SpindleSource struct { + Spindle string +} + +func (s SpindleSource) Key() string { + return s.Spindle +} + +func (s SpindleSource) Url(cursor int64, dev bool) (*url.URL, error) { + scheme := "wss" + if dev { + scheme = "ws" + } + + u, err := url.Parse(scheme + "://" + s.Spindle + "/events") + if err != nil { + return nil, err + } + + if cursor != 0 { + query := url.Values{} + query.Add("cursor", fmt.Sprintf("%d", cursor)) + u.RawQuery = query.Encode() + } + return u, nil +} + +func NewSpindleSource(spindle string) SpindleSource { + return SpindleSource{ + Spindle: spindle, + } +} diff --git a/knotclient/events.go b/knotclient/events.go deleted file mode 100644 --- a/knotclient/events.go +++ /dev/null @@ -1,268 +0,0 @@ -package knotclient - -import ( - "context" - "encoding/json" - "fmt" - "log/slog" - "math/rand" - "net/url" - "sync" - "time" - - "tangled.sh/tangled.sh/core/knotclient/cursor" - "tangled.sh/tangled.sh/core/log" - - "github.com/gorilla/websocket" -) - -type ProcessFunc func(ctx context.Context, source EventSource, message Message) error - -type Message struct { - Rkey string - Nsid string - // do not full deserialize this portion of the message, processFunc can do that - EventJson json.RawMessage `json:"event"` -} - -type ConsumerConfig struct { - Sources map[EventSource]struct{} - ProcessFunc ProcessFunc - RetryInterval time.Duration - MaxRetryInterval time.Duration - ConnectionTimeout time.Duration - WorkerCount int - QueueSize int - Logger *slog.Logger - Dev bool - CursorStore cursor.Store -} - -func NewConsumerConfig() *ConsumerConfig { - return &ConsumerConfig{ - Sources: make(map[EventSource]struct{}), - } -} - -type EventSource struct { - Knot string -} - -func NewEventSource(knot string) EventSource { - return EventSource{ - Knot: knot, - } -} - -type EventConsumer struct { - wg sync.WaitGroup - dialer *websocket.Dialer - connMap sync.Map - jobQueue chan job - logger *slog.Logger - randSource *rand.Rand - - // rw lock over edits to ConsumerConfig - cfgMu sync.RWMutex - cfg ConsumerConfig -} - -func (e *EventConsumer) buildUrl(s EventSource, cursor int64) (*url.URL, error) { - scheme := "wss" - if e.cfg.Dev { - scheme = "ws" - } - - u, err := url.Parse(scheme + "://" + s.Knot + "/events") - if err != nil { - return nil, err - } - - if cursor != 0 { - query := url.Values{} - query.Add("cursor", fmt.Sprintf("%d", cursor)) - u.RawQuery = query.Encode() - } - return u, nil -} - -type job struct { - source EventSource - message []byte -} - -func NewEventConsumer(cfg ConsumerConfig) *EventConsumer { - if cfg.RetryInterval == 0 { - cfg.RetryInterval = 15 * time.Minute - } - if cfg.ConnectionTimeout == 0 { - cfg.ConnectionTimeout = 10 * time.Second - } - if cfg.WorkerCount <= 0 { - cfg.WorkerCount = 5 - } - if cfg.MaxRetryInterval == 0 { - cfg.MaxRetryInterval = 1 * time.Hour - } - if cfg.Logger == nil { - cfg.Logger = log.New("eventconsumer") - } - if cfg.QueueSize == 0 { - cfg.QueueSize = 100 - } - if cfg.CursorStore == nil { - cfg.CursorStore = &cursor.MemoryStore{} - } - return &EventConsumer{ - cfg: cfg, - dialer: websocket.DefaultDialer, - jobQueue: make(chan job, cfg.QueueSize), // buffered job queue - logger: cfg.Logger, - randSource: rand.New(rand.NewSource(time.Now().UnixNano())), - } -} - -func (c *EventConsumer) Start(ctx context.Context) { - c.cfg.Logger.Info("starting consumer", "config", c.cfg) - - // start workers - for range c.cfg.WorkerCount { - c.wg.Add(1) - go c.worker(ctx) - } - - // start streaming - for source := range c.cfg.Sources { - c.wg.Add(1) - go c.startConnectionLoop(ctx, source) - } -} - -func (c *EventConsumer) Stop() { - c.connMap.Range(func(_, val any) bool { - if conn, ok := val.(*websocket.Conn); ok { - conn.Close() - } - return true - }) - c.wg.Wait() - close(c.jobQueue) -} - -func (c *EventConsumer) AddSource(ctx context.Context, s EventSource) { - // we are already listening to this source - if _, ok := c.cfg.Sources[s]; ok { - c.logger.Info("source already present", "source", s) - return - } - - c.cfgMu.Lock() - c.cfg.Sources[s] = struct{}{} - c.wg.Add(1) - go c.startConnectionLoop(ctx, s) - c.cfgMu.Unlock() -} - -func (c *EventConsumer) worker(ctx context.Context) { - defer c.wg.Done() - for { - select { - case <-ctx.Done(): - return - case j, ok := <-c.jobQueue: - if !ok { - return - } - - var msg Message - err := json.Unmarshal(j.message, &msg) - if err != nil { - c.logger.Error("error deserializing message", "source", j.source.Knot, "err", err) - return - } - - // update cursor - c.cfg.CursorStore.Set(j.source.Knot, time.Now().UnixNano()) - - if err := c.cfg.ProcessFunc(ctx, j.source, msg); err != nil { - c.logger.Error("error processing message", "source", j.source, "err", err) - } - } - } -} - -func (c *EventConsumer) startConnectionLoop(ctx context.Context, source EventSource) { - defer c.wg.Done() - retryInterval := c.cfg.RetryInterval - for { - select { - case <-ctx.Done(): - return - default: - err := c.runConnection(ctx, source) - if err != nil { - c.logger.Error("connection failed", "source", source, "err", err) - } - - // apply jitter - jitter := time.Duration(c.randSource.Int63n(int64(retryInterval) / 5)) - delay := retryInterval + jitter - - if retryInterval < c.cfg.MaxRetryInterval { - retryInterval *= 2 - if retryInterval > c.cfg.MaxRetryInterval { - retryInterval = c.cfg.MaxRetryInterval - } - } - c.logger.Info("retrying connection", "source", source, "delay", delay) - select { - case <-time.After(delay): - case <-ctx.Done(): - return - } - } - } -} - -func (c *EventConsumer) runConnection(ctx context.Context, source EventSource) error { - connCtx, cancel := context.WithTimeout(ctx, c.cfg.ConnectionTimeout) - defer cancel() - - cursor := c.cfg.CursorStore.Get(source.Knot) - - u, err := c.buildUrl(source, cursor) - if err != nil { - return err - } - - c.logger.Info("connecting", "url", u.String()) - conn, _, err := c.dialer.DialContext(connCtx, u.String(), nil) - if err != nil { - return err - } - defer conn.Close() - c.connMap.Store(source, conn) - defer c.connMap.Delete(source) - - c.logger.Info("connected", "source", source) - - for { - select { - case <-ctx.Done(): - return nil - default: - msgType, msg, err := conn.ReadMessage() - if err != nil { - return err - } - if msgType != websocket.TextMessage { - continue - } - select { - case c.jobQueue <- job{source: source, message: msg}: - case <-ctx.Done(): - return nil - } - } - } -} diff --git a/spindle/ingester.go b/spindle/ingester.go --- a/spindle/ingester.go +++ b/spindle/ingester.go @@ -6,7 +6,7 @@ "fmt" "tangled.sh/tangled.sh/core/api/tangled" - "tangled.sh/tangled.sh/core/knotclient" + "tangled.sh/tangled.sh/core/eventconsumer" "github.com/bluesky-social/jetstream/pkg/models" ) @@ -128,7 +128,7 @@ } // add this knot to the event consumer - src := knotclient.NewEventSource(record.Knot) + src := eventconsumer.NewKnotSource(record.Knot) s.ks.AddSource(context.Background(), src) return nil diff --git a/spindle/server.go b/spindle/server.go --- a/spindle/server.go +++ b/spindle/server.go @@ -9,9 +9,9 @@ "github.com/go-chi/chi/v5" "tangled.sh/tangled.sh/core/api/tangled" + "tangled.sh/tangled.sh/core/eventconsumer" + "tangled.sh/tangled.sh/core/eventconsumer/cursor" "tangled.sh/tangled.sh/core/jetstream" - "tangled.sh/tangled.sh/core/knotclient" - "tangled.sh/tangled.sh/core/knotclient/cursor" "tangled.sh/tangled.sh/core/log" "tangled.sh/tangled.sh/core/notifier" "tangled.sh/tangled.sh/core/rbac" @@ -35,7 +35,7 @@ eng *engine.Engine jq *queue.Queue cfg *config.Config - ks *knotclient.EventConsumer + ks *eventconsumer.Consumer } func Run(ctx context.Context) error { @@ -114,7 +114,7 @@ // for each incoming sh.tangled.pipeline, we execute // spindle.processPipeline, which in turn enqueues the pipeline // job in the above registered queue. - ccfg := knotclient.NewConsumerConfig() + ccfg := eventconsumer.NewConsumerConfig() ccfg.Logger = logger ccfg.Dev = cfg.Server.Dev ccfg.ProcessFunc = spindle.processPipeline @@ -125,9 +125,9 @@ } for _, knot := range knownKnots { logger.Info("adding source start", "knot", knot) - ccfg.Sources[knotclient.EventSource{knot}] = struct{}{} + ccfg.Sources[eventconsumer.NewKnotSource(knot)] = struct{}{} } - spindle.ks = knotclient.NewEventConsumer(*ccfg) + spindle.ks = eventconsumer.NewConsumer(*ccfg) go func() { logger.Info("starting knot event consumer") @@ -151,7 +151,7 @@ return mux } -func (s *Spindle) processPipeline(ctx context.Context, src knotclient.EventSource, msg knotclient.Message) error { +func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, msg eventconsumer.Message) error { if msg.Nsid == tangled.PipelineNSID { pipeline := tangled.Pipeline{} err := json.Unmarshal(msg.EventJson, &pipeline) @@ -179,7 +179,7 @@ } pipelineId := models.PipelineId{ - Knot: src.Knot, + Knot: src.Key(), Rkey: msg.Rkey, } diff --git a/spindle/stream.go b/spindle/stream.go --- a/spindle/stream.go +++ b/spindle/stream.go @@ -2,6 +2,7 @@ import ( "context" + "encoding/json" "fmt" "net/http" "strconv" @@ -206,19 +207,37 @@ } func (s *Spindle) streamPipelines(conn *websocket.Conn, cursor *int64) error { - ops, err := s.db.GetEvents(*cursor) + events, err := s.db.GetEvents(*cursor) if err != nil { s.l.Debug("err", "err", err) return err } - s.l.Debug("ops", "ops", ops) + s.l.Debug("ops", "ops", events) - for _, op := range ops { - if err := conn.WriteJSON(op); err != nil { + for _, event := range events { + // first extract the inner json into a map + var eventJson map[string]any + err := json.Unmarshal([]byte(event.EventJson), &eventJson) + if err != nil { + s.l.Error("failed to unmarshal event", "err", err) + return err + } + + jsonMsg, err := json.Marshal(map[string]any{ + "rkey": event.Rkey, + "nsid": event.Nsid, + "event": eventJson, + }) + if err != nil { + s.l.Error("failed to marshal record", "err", err) + return err + } + + if err := conn.WriteMessage(websocket.TextMessage, jsonMsg); err != nil { s.l.Debug("err", "err", err) return err } - *cursor = op.Created + *cursor = event.Created } return nil diff --git a/eventconsumer/cursor/memory.go b/eventconsumer/cursor/memory.go new file mode 100644 --- /dev/null +++ b/eventconsumer/cursor/memory.go @@ -0,0 +1,23 @@ +package cursor + +import ( + "sync" +) + +type MemoryStore struct { + store sync.Map +} + +func (m *MemoryStore) Set(knot string, cursor int64) { + m.store.Store(knot, cursor) +} + +func (m *MemoryStore) Get(knot string) (cursor int64) { + if result, ok := m.store.Load(knot); ok { + if val, ok := result.(int64); ok { + return val + } + } + + return 0 +} diff --git a/eventconsumer/cursor/redis.go b/eventconsumer/cursor/redis.go new file mode 100644 --- /dev/null +++ b/eventconsumer/cursor/redis.go @@ -0,0 +1,43 @@ +package cursor + +import ( + "context" + "fmt" + "strconv" + + "tangled.sh/tangled.sh/core/appview/cache" +) + +const ( + cursorKey = "cursor:%s" +) + +type RedisStore struct { + rdb *cache.Cache +} + +func NewRedisCursorStore(cache *cache.Cache) RedisStore { + return RedisStore{ + rdb: cache, + } +} + +func (r *RedisStore) Set(knot string, cursor int64) { + key := fmt.Sprintf(cursorKey, knot) + r.rdb.Set(context.Background(), key, cursor, 0) +} + +func (r *RedisStore) Get(knot string) (cursor int64) { + key := fmt.Sprintf(cursorKey, knot) + val, err := r.rdb.Get(context.Background(), key).Result() + if err != nil { + return 0 + } + cursor, err = strconv.ParseInt(val, 10, 64) + if err != nil { + // TODO: log here + return 0 + } + + return cursor +} diff --git a/eventconsumer/cursor/sqlite.go b/eventconsumer/cursor/sqlite.go new file mode 100644 --- /dev/null +++ b/eventconsumer/cursor/sqlite.go @@ -0,0 +1,83 @@ +package cursor + +import ( + "database/sql" + "fmt" + + _ "github.com/mattn/go-sqlite3" +) + +type SqliteStore struct { + db *sql.DB + tableName string +} + +type SqliteStoreOpt func(*SqliteStore) + +func WithTableName(name string) SqliteStoreOpt { + return func(s *SqliteStore) { + s.tableName = name + } +} + +func NewSQLiteStore(dbPath string, opts ...SqliteStoreOpt) (*SqliteStore, error) { + db, err := sql.Open("sqlite3", dbPath) + if err != nil { + return nil, fmt.Errorf("failed to open sqlite database: %w", err) + } + + store := &SqliteStore{ + db: db, + tableName: "cursors", + } + + for _, o := range opts { + o(store) + } + + if err := store.init(); err != nil { + return nil, err + } + + return store, nil +} + +func (s *SqliteStore) init() error { + createTable := fmt.Sprintf(` + create table if not exists %s ( + knot text primary key, + cursor text + );`, s.tableName) + _, err := s.db.Exec(createTable) + return err +} + +func (s *SqliteStore) Set(knot string, cursor int64) { + query := fmt.Sprintf(` + insert into %s (knot, cursor) + values (?, ?) + on conflict(knot) do update set cursor=excluded.cursor; + `, s.tableName) + + _, err := s.db.Exec(query, knot, cursor) + + if err != nil { + // TODO: log here + } +} + +func (s *SqliteStore) Get(knot string) (cursor int64) { + query := fmt.Sprintf(` + select cursor from %s where knot = ?; + `, s.tableName) + err := s.db.QueryRow(query, knot).Scan(&cursor) + + if err != nil { + if err != sql.ErrNoRows { + // TODO: log here + } + return 0 + } + + return cursor +} diff --git a/eventconsumer/cursor/store.go b/eventconsumer/cursor/store.go new file mode 100644 --- /dev/null +++ b/eventconsumer/cursor/store.go @@ -0,0 +1,6 @@ +package cursor + +type Store interface { + Set(knot string, cursor int64) + Get(knot string) (cursor int64) +} diff --git a/knotclient/cursor/memory.go b/knotclient/cursor/memory.go deleted file mode 100644 --- a/knotclient/cursor/memory.go +++ /dev/null @@ -1,23 +0,0 @@ -package cursor - -import ( - "sync" -) - -type MemoryStore struct { - store sync.Map -} - -func (m *MemoryStore) Set(knot string, cursor int64) { - m.store.Store(knot, cursor) -} - -func (m *MemoryStore) Get(knot string) (cursor int64) { - if result, ok := m.store.Load(knot); ok { - if val, ok := result.(int64); ok { - return val - } - } - - return 0 -} diff --git a/knotclient/cursor/redis.go b/knotclient/cursor/redis.go deleted file mode 100644 --- a/knotclient/cursor/redis.go +++ /dev/null @@ -1,43 +0,0 @@ -package cursor - -import ( - "context" - "fmt" - "strconv" - - "tangled.sh/tangled.sh/core/appview/cache" -) - -const ( - cursorKey = "cursor:%s" -) - -type RedisStore struct { - rdb *cache.Cache -} - -func NewRedisCursorStore(cache *cache.Cache) RedisStore { - return RedisStore{ - rdb: cache, - } -} - -func (r *RedisStore) Set(knot string, cursor int64) { - key := fmt.Sprintf(cursorKey, knot) - r.rdb.Set(context.Background(), key, cursor, 0) -} - -func (r *RedisStore) Get(knot string) (cursor int64) { - key := fmt.Sprintf(cursorKey, knot) - val, err := r.rdb.Get(context.Background(), key).Result() - if err != nil { - return 0 - } - cursor, err = strconv.ParseInt(val, 10, 64) - if err != nil { - // TODO: log here - return 0 - } - - return cursor -} diff --git a/knotclient/cursor/sqlite.go b/knotclient/cursor/sqlite.go deleted file mode 100644 --- a/knotclient/cursor/sqlite.go +++ /dev/null @@ -1,83 +0,0 @@ -package cursor - -import ( - "database/sql" - "fmt" - - _ "github.com/mattn/go-sqlite3" -) - -type SqliteStore struct { - db *sql.DB - tableName string -} - -type SqliteStoreOpt func(*SqliteStore) - -func WithTableName(name string) SqliteStoreOpt { - return func(s *SqliteStore) { - s.tableName = name - } -} - -func NewSQLiteStore(dbPath string, opts ...SqliteStoreOpt) (*SqliteStore, error) { - db, err := sql.Open("sqlite3", dbPath) - if err != nil { - return nil, fmt.Errorf("failed to open sqlite database: %w", err) - } - - store := &SqliteStore{ - db: db, - tableName: "cursors", - } - - for _, o := range opts { - o(store) - } - - if err := store.init(); err != nil { - return nil, err - } - - return store, nil -} - -func (s *SqliteStore) init() error { - createTable := fmt.Sprintf(` - create table if not exists %s ( - knot text primary key, - cursor text - );`, s.tableName) - _, err := s.db.Exec(createTable) - return err -} - -func (s *SqliteStore) Set(knot string, cursor int64) { - query := fmt.Sprintf(` - insert into %s (knot, cursor) - values (?, ?) - on conflict(knot) do update set cursor=excluded.cursor; - `, s.tableName) - - _, err := s.db.Exec(query, knot, cursor) - - if err != nil { - // TODO: log here - } -} - -func (s *SqliteStore) Get(knot string) (cursor int64) { - query := fmt.Sprintf(` - select cursor from %s where knot = ?; - `, s.tableName) - err := s.db.QueryRow(query, knot).Scan(&cursor) - - if err != nil { - if err != sql.ErrNoRows { - // TODO: log here - } - return 0 - } - - return cursor -} diff --git a/knotclient/cursor/store.go b/knotclient/cursor/store.go deleted file mode 100644 --- a/knotclient/cursor/store.go +++ /dev/null @@ -1,6 +0,0 @@ -package cursor - -type Store interface { - Set(knot string, cursor int64) - Get(knot string) (cursor int64) -} -- tangled.sh