diff --git a/go.mod b/go.mod index 7da1f11..2a4019d 100644 --- a/go.mod +++ b/go.mod @@ -30,12 +30,14 @@ require ( 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 ) @@ -50,6 +52,7 @@ require ( 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 @@ -102,6 +105,7 @@ require ( 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 diff --git a/knotserver/db/events.go b/knotserver/db/events.go new file mode 100644 index 0000000..e67f82a --- /dev/null +++ b/knotserver/db/events.go @@ -0,0 +1,62 @@ +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 +} diff --git a/knotserver/db/init.go b/knotserver/db/init.go index ec7233b..b275233 100644 --- a/knotserver/db/init.go +++ b/knotserver/db/init.go @@ -44,13 +44,11 @@ func Setup(dbPath string) (*DB, error) { 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 { diff --git a/knotserver/db/oplog.go b/knotserver/db/oplog.go deleted file mode 100644 index 842886c..0000000 --- a/knotserver/db/oplog.go +++ /dev/null @@ -1,70 +0,0 @@ -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 // 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 -} diff --git a/knotserver/events.go b/knotserver/events.go index 7bbf39b..7cc7f3a 100644 --- a/knotserver/events.go +++ b/knotserver/events.go @@ -2,6 +2,7 @@ package knotserver import ( "context" + "encoding/json" "net/http" "time" @@ -13,7 +14,7 @@ var upgrader = websocket.Upgrader{ 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") @@ -74,19 +75,37 @@ func (h *Handle) OpLog(w http.ResponseWriter, r *http.Request) { } 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 diff --git a/knotserver/handler.go b/knotserver/handler.go index 48d6462..27142c0 100644 --- a/knotserver/handler.go +++ b/knotserver/handler.go @@ -149,7 +149,7 @@ func Setup(ctx context.Context, c *config.Config, db *db.DB, e *rbac.Enforcer, j }) // 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) diff --git a/knotserver/internal.go b/knotserver/internal.go index 610fb8a..39c0687 100644 --- a/knotserver/internal.go +++ b/knotserver/internal.go @@ -2,12 +2,14 @@ package knotserver 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" @@ -67,6 +69,15 @@ func (h *InternalHandle) PostReceiveHook(w http.ResponseWriter, r *http.Request) 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) @@ -76,25 +87,35 @@ func (h *InternalHandle) PostReceiveHook(w http.ResponseWriter, r *http.Request) } 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) 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, +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, } - return h.db.InsertOp(op, h.n) + eventJson, err := json.Marshal(refUpdate) + if err != nil { + return err + } + + event := db.Event{ + Rkey: TID(), + Nsid: tangled.GitRefUpdateNSID, + EventJson: string(eventJson), + } + + 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 {