knotserver: replace oplog with generic event store

the `db.Op` event is now replaced by the `refUpdate` event. all knot
generated events will be stored in the events db as raw json.

Signed-off-by: oppiliappan <[e-mail]>

go.mod

···
 	github.com/posthog/posthog-go v1.5.5
 	github.com/resend/resend-go/v2 v2.15.0
 	github.com/sethvargo/go-envconfig v1.1.0
+	github.com/stretchr/testify v1.10.0
 	github.com/urfave/cli/v3 v3.3.3
 	github.com/whyrusleeping/cbor-gen v0.3.1
 	github.com/yuin/goldmark v1.4.13
 	golang.org/x/crypto v0.38.0
 	golang.org/x/net v0.39.0
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da
+	gopkg.in/yaml.v3 v3.0.1
 	tangled.sh/icyphox.sh/atproto-oauth v0.0.0-20250526154904-3906c5336421
 )
 
···
 	github.com/casbin/govaluate v1.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudflare/circl v1.6.0 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/dlclark/regexp2 v1.11.5 // indirect
···
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pjbgf/sha1cd v0.3.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/polydawn/refmt v0.89.1-0.20221221234430-40501e09de1f // indirect
 	github.com/prometheus/client_golang v1.22.0 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect

knotserver/db/events.go

···
+package db
+
+import (
+	"fmt"
+
+	"tangled.sh/tangled.sh/core/knotserver/notifier"
+)
+
+type Event struct {
+	Rkey      string `json:"rkey"`
+	Nsid      string `json:"nsid"`
+	EventJson string `json:"event"`
+}
+
+func (d *DB) InsertEvent(event Event, notifier *notifier.Notifier) error {
+	_, err := d.db.Exec(
+		`insert into events (rkey, nsid, event) values (?, ?, ?)`,
+		event.Rkey,
+		event.Nsid,
+		event.EventJson,
+	)
+
+	notifier.NotifyAll()
+
+	return err
+}
+
+func (d *DB) GetEvents(cursor string) ([]Event, error) {
+	whereClause := ""
+	args := []any{}
+	if cursor != "" {
+		whereClause = "where rkey > ?"
+		args = append(args, cursor)
+	}
+
+	query := fmt.Sprintf(`
+		select rkey, nsid, event
+		from events
+		%s
+		order by rkey asc
+		limit 100
+	`, whereClause)
+
+	rows, err := d.db.Query(query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var evts []Event
+	for rows.Next() {
+		var ev Event
+		rows.Scan(&ev.Rkey, &ev.Nsid, &ev.EventJson)
+		evts = append(evts, ev)
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	return evts, nil
+}

knotserver/db/init.go

···
 			last_time_us integer not null
 		);
 
-		create table if not exists oplog (
-			tid     text primary key,
-			did     text not null,
-			repo    text not null,
-			old_sha text not null,
-			new_sha text not null,
-			ref     text not null
+		create table if not exists events (
+			rkey text not null,
+			nsid text not null,
+			event text not null, -- json
+			primary key (rkey, nsid)
 		);
 	`)
 	if err != nil {

knotserver/db/oplog.go

···
-package db
-
-import (
-	"fmt"
-
-	"tangled.sh/tangled.sh/core/knotserver/notifier"
-)
-
-type Op struct {
-	Tid    string // time based ID, easy to enumerate & monotonic
-	Did    string // did of pusher
-	Repo   string // <did/repo> fully qualified repo
-	OldSha string // old sha of reference being updated
-	NewSha string // new sha of reference being updated
-	Ref    string // the reference being updated
-}
-
-func (d *DB) InsertOp(op Op, notifier *notifier.Notifier) error {
-	_, err := d.db.Exec(
-		`insert into oplog (tid, did, repo, old_sha, new_sha, ref) values (?, ?, ?, ?, ?, ?)`,
-		op.Tid,
-		op.Did,
-		op.Repo,
-		op.OldSha,
-		op.NewSha,
-		op.Ref,
-	)
-	if err != nil {
-		return err
-	}
-
-	notifier.NotifyAll()
-	return nil
-}
-
-func (d *DB) GetOps(cursor string) ([]Op, error) {
-	whereClause := ""
-	args := []any{}
-	if cursor != "" {
-		whereClause = "where tid > ?"
-		args = append(args, cursor)
-	}
-
-	query := fmt.Sprintf(`
-		select tid, did, repo, old_sha, new_sha, ref
-		from oplog
-		%s
-		order by tid asc
-		limit 100
-	`, whereClause)
-
-	rows, err := d.db.Query(query, args...)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-
-	var ops []Op
-	for rows.Next() {
-		var op Op
-		rows.Scan(&op.Tid, &op.Did, &op.Repo, &op.OldSha, &op.NewSha, &op.Ref)
-		ops = append(ops, op)
-	}
-
-	if err := rows.Err(); err != nil {
-		return nil, err
-	}
-
-	return ops, nil
-}

knotserver/events.go

···
 
 import (
 	"context"
+	"encoding/json"
 	"net/http"
 	"time"
 
···
 	WriteBufferSize: 1024,
 }
 
-func (h *Handle) OpLog(w http.ResponseWriter, r *http.Request) {
+func (h *Handle) Events(w http.ResponseWriter, r *http.Request) {
 	l := h.l.With("handler", "OpLog")
 	l.Info("received new connection")
 
···
 }
 
 func (h *Handle) streamOps(conn *websocket.Conn, cursor *string) error {
-	ops, err := h.db.GetOps(*cursor)
+	events, err := h.db.GetEvents(*cursor)
 	if err != nil {
-		h.l.Debug("err", "err", err)
+		h.l.Error("failed to fetch events from db", "err", err, "cursor", cursor)
 		return err
 	}
-	h.l.Debug("ops", "ops", ops)
+	h.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 {
+			h.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 {
+			h.l.Error("failed to marshal record", "err", err)
+			return err
+		}
+
+		if err := conn.WriteMessage(websocket.TextMessage, jsonMsg); err != nil {
 			h.l.Debug("err", "err", err)
 			return err
 		}
-		*cursor = op.Tid
+		*cursor = event.Rkey
 	}
 
 	return nil

knotserver/handler.go

···
 	})
 
 	// Socket that streams git oplogs
-	r.Get("/oplog", h.OpLog)
+	r.Get("/events", h.Events)
 
 	// Initialize the knot with an owner and public key.
 	r.With(h.VerifySignature).Post("/init", h.Init)

knotserver/internal.go

···
 
 import (
 	"context"
+	"encoding/json"
 	"log/slog"
 	"net/http"
 	"path/filepath"
 
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/chi/v5/middleware"
+	"tangled.sh/tangled.sh/core/api/tangled"
 	"tangled.sh/tangled.sh/core/knotserver/config"
 	"tangled.sh/tangled.sh/core/knotserver/db"
 	"tangled.sh/tangled.sh/core/knotserver/git"
···
 		l.Error("failed to calculate relative git dir", "scanPath", h.c.Repo.ScanPath, "gitAbsoluteDir", gitAbsoluteDir)
 		return
 	}
+
+	parts := strings.SplitN(gitRelativeDir, "/", 2)
+	if len(parts) != 2 {
+		l.Error("invalid git dir", "gitRelativeDir", gitRelativeDir)
+		return
+	}
+	repoDid := parts[0]
+	repoName := parts[1]
+
 	gitUserDid := r.Header.Get("X-Git-User-Did")
 
 	lines, err := git.ParsePostReceive(r.Body)
···
 	}
 
 	for _, line := range lines {
-		err := h.updateOpLog(line, gitUserDid, gitRelativeDir)
+		err := h.insertRefUpdate(line, gitUserDid, repoDid, repoName)
 		if err != nil {
 			l.Error("failed to insert op", "err", err, "line", line, "did", gitUserDid, "repo", gitRelativeDir)
+			// non-fatal
 		}
 	}
+}
 
-	return
-}
+func (h *InternalHandle) insertRefUpdate(line git.PostReceiveLine, gitUserDid, repoDid, repoName string) error {
+	refUpdate := tangled.GitRefUpdate{
+		OldSha:       line.OldSha,
+		NewSha:       line.NewSha,
+		Ref:          line.Ref,
+		CommitterDid: gitUserDid,
+		RepoDid:      repoDid,
+		RepoName:     repoName,
+	}
+	eventJson, err := json.Marshal(refUpdate)
+	if err != nil {
+		return err
+	}
 
-func (h *InternalHandle) updateOpLog(line git.PostReceiveLine, did, repo string) error {
-	op := db.Op{
-		Tid:    TID(),
-		Did:    did,
-		Repo:   repo,
-		OldSha: line.OldSha,
-		NewSha: line.NewSha,
-		Ref:    line.Ref,
+	event := db.Event{
+		Rkey:      TID(),
+		Nsid:      tangled.GitRefUpdateNSID,
+		EventJson: string(eventJson),
 	}
-	return h.db.InsertOp(op, h.n)
+
+	return h.db.InsertEvent(event, h.n)
 }
 
 func Internal(ctx context.Context, c *config.Config, db *db.DB, e *rbac.Enforcer, l *slog.Logger, n *notifier.Notifier) http.Handler {