From d8a5164f5f1c6ac49b056eb0cb1b9c3d9db338fd Mon Sep 17 00:00:00 2001 From: Eugene Blikh Date: Fri, 24 Jul 2026 12:49:46 +0300 Subject: [PATCH] feat(webhooks): fire on proposal open/merge/reject (Phase 5a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The firing half — proposal lifecycle events now deliver GraphQL-native webhooks. Verified end to end against a live daemon: an agent REST propose delivers a signed POST whose body is the subscription's stored query executed against the ProposalEvent payload. - service: an EventSink seam (service/events.go). Propose emits PROPOSAL_OPENED for a new proposal, mergeProposal emits PROPOSAL_MERGED (the single merge point — both auto-merge and the human approve reach it), Reject emits PROPOSAL_REJECTED. Nil-safe; a Service with no sink emits nothing. - graph.NewProposalEvent builds the *model.ProposalEvent payload from a service.Proposal (reusing the existing service→graph→model mapping). - cmd webhookEventSink: proposal events happen in the service layer, which has none of core-go's request context, so the sink enqueues a dowork task onto the webhook queue. The task runs in the queue's worker context (server+database+config, from WithQueues), adds the owner's INTERNAL auth, and calls Schedule — which renders each subscriber's query and delivers it Ed25519-signed. Fire-and-forget off the write path: a webhook never blocks or fails a proposal write. Phase 5a (webhooks) is complete: DB, the authn→AuthContext bridge, the GraphQL surface, the core-go server wiring, and firing. --- cmd/specsrht/main.go | 6 ++ cmd/specsrht/webhooks.go | 59 ++++++++++++ go.mod | 2 +- graph/webhook_event.go | 45 +++++++++ service/events.go | 18 ++++ service/events_test.go | 200 +++++++++++++++++++++++++++++++++++++++ service/merge.go | 16 +++- service/propose.go | 8 ++ service/service.go | 15 +++ 9 files changed, 366 insertions(+), 3 deletions(-) create mode 100644 cmd/specsrht/webhooks.go create mode 100644 graph/webhook_event.go create mode 100644 service/events.go create mode 100644 service/events_test.go diff --git a/cmd/specsrht/main.go b/cmd/specsrht/main.go index db67e09098234260db9df14a12e9c52bb8810114..0b8849f1687ac3e59d997aa900d2910086fec3bc 100644 --- a/cmd/specsrht/main.go +++ b/cmd/specsrht/main.go @@ -314,6 +314,12 @@ func run(log *slog.Logger) error { WithQueues(webhookQueue.Queue) mountRoutes(srv.AnonRouter(), conf, surf) + // Now that the webhook queue is started (WithQueues gave its worker the + // server+database+config context), install the sink so proposal lifecycle + // events fire deliveries. The owner user id is valid — EnsureOwnerUser ran + // above. + svc.SetEventSink(newWebhookEventSink(webhookQueue, svc.OwnerUserID(), cfg.Instance.OwnerName, log)) + ctx, stop := context.WithCancel(context.Background()) defer stop() diff --git a/cmd/specsrht/webhooks.go b/cmd/specsrht/webhooks.go new file mode 100644 index 0000000000000000000000000000000000000000..c674a8c3c893c01c9c0c16fa87b17a2cac9fad98 --- /dev/null +++ b/cmd/specsrht/webhooks.go @@ -0,0 +1,59 @@ +package main + +import ( + "context" + "log/slog" + "time" + + work "git.sr.ht/~sircmpwn/dowork" + sq "github.com/Masterminds/squirrel" + "github.com/google/uuid" + + "sourcecraft.dev/bigbes/sr-ht-core/auth" + "sourcecraft.dev/bigbes/sr-ht-core/webhooks" + + "sourcecraft.dev/bigbes/sr-ht-spec/graph" + "sourcecraft.dev/bigbes/sr-ht-spec/graph/model" + "sourcecraft.dev/bigbes/sr-ht-spec/service" +) + +// webhookEventSink implements service.EventSink by enqueuing a dowork task onto +// the webhook queue. The task runs in the queue's worker context (server + +// database + config, from WithQueues), adds the owner's INTERNAL auth context, +// and calls Schedule — which needs all three. Firing is fire-and-forget: a +// webhook must never block or fail a proposal write. +type webhookEventSink struct { + queue *webhooks.WebhookQueue + ownerUserID int + ownerName string + log *slog.Logger +} + +func newWebhookEventSink(q *webhooks.WebhookQueue, ownerUserID int, ownerName string, log *slog.Logger) *webhookEventSink { + return &webhookEventSink{queue: q, ownerUserID: ownerUserID, ownerName: ownerName, log: log} +} + +// Compile-time assertion that the sink satisfies the service seam. +var _ service.EventSink = (*webhookEventSink)(nil) + +func (s *webhookEventSink) ProposalEvent(kind service.ProposalEventKind, p service.Proposal) { + u := uuid.New() + payload, err := graph.NewProposalEvent(model.WebhookEvent(kind), u.String(), time.Now().UTC(), p) + if err != nil { + s.log.Error("build webhook payload", "err", err) + return + } + event := string(kind) + task := work.NewTask(func(ctx context.Context) error { + // The worker context carries server+database+config; add the owner auth + // Schedule captures (fetchSubscriptions and the delivery both need it). + ctx = auth.Context(ctx, &auth.AuthContext{ + AuthMethod: auth.AUTH_INTERNAL, UserID: s.ownerUserID, Username: s.ownerName, + }) + q := sq.Select().From("gql_user_wh_sub sub").Where("sub.user_id = ?", s.ownerUserID) + s.queue.Schedule(ctx, q, "user", event, u, payload) + return nil + }) + // Enqueue off the write path so a full queue never blocks a merge. + go func() { _ = s.queue.Queue.Enqueue(task) }() +} diff --git a/go.mod b/go.mod index 2d23af51fd2109b862a2703ac190bea89db98559..65eccfd2c7bf940915d8c6fc0304223e2086834d 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.26.4 require ( git.sr.ht/~bitfehler/brant v0.5.1 + git.sr.ht/~sircmpwn/dowork v0.0.0-20241216125407-2b00aa42322c github.com/99designs/gqlgen v0.17.36 github.com/Masterminds/squirrel v1.5.4 github.com/alexflint/go-arg v1.6.0 @@ -25,7 +26,6 @@ require ( require ( dario.cat/mergo v1.0.0 // indirect - git.sr.ht/~sircmpwn/dowork v0.0.0-20241216125407-2b00aa42322c // indirect git.sr.ht/~sircmpwn/getopt v1.0.0 // indirect git.sr.ht/~sircmpwn/go-bare v0.0.0-20210406120253-ab86bc2846d9 // indirect github.com/Microsoft/go-winio v0.6.2 // indirect diff --git a/graph/webhook_event.go b/graph/webhook_event.go new file mode 100644 index 0000000000000000000000000000000000000000..c4536e05585d1d4a9ee971774574bfc528c6d1fd --- /dev/null +++ b/graph/webhook_event.go @@ -0,0 +1,45 @@ +package graph + +import ( + "time" + + "sourcecraft.dev/bigbes/sr-ht-spec/graph/model" + "sourcecraft.dev/bigbes/sr-ht-spec/service" +) + +// NewProposalEvent builds the webhook payload for a proposal lifecycle event. +// The cmd webhook sink calls it; it lives here because converting a +// service.Proposal to the GraphQL model.Proposal is this package's job. +// +// It composes the two existing conversions this package already owns: a +// service.Proposal is field-for-field a graph.Proposal (the same shape +// serviceProposals.ListProposals produces), and proposalModel maps a +// graph.Proposal to the *model.Proposal every GraphQL surface returns. Reusing +// them keeps the webhook payload identical to what a `proposals` query would +// serve for the same proposal. +func NewProposalEvent(event model.WebhookEvent, uuid string, date time.Time, p service.Proposal) (*model.ProposalEvent, error) { + pm, err := proposalModel(Proposal{ + ID: p.ID, + Space: p.Space, + Title: p.Title, + Rationale: p.Rationale, + BaseRev: p.BaseRev, + Branch: p.Branch, + State: p.State, + Approval: p.Approval, + MergedRev: p.MergedRev, + Agent: p.Agent, + AgentSession: p.AgentSession, + Created: p.Created, + Resolved: p.Resolved, + }) + if err != nil { + return nil, err + } + return &model.ProposalEvent{ + UUID: uuid, + Event: event, + Date: date, + Proposal: pm, + }, nil +} diff --git a/service/events.go b/service/events.go new file mode 100644 index 0000000000000000000000000000000000000000..dc8cace6b33cd91ccbd4d425eaa61a49b75861f3 --- /dev/null +++ b/service/events.go @@ -0,0 +1,18 @@ +package service + +// ProposalEventKind names a proposal lifecycle event. The values equal the +// graph WebhookEvent enum values so the surface maps them without a table. +type ProposalEventKind string + +const ( + EventProposalOpened ProposalEventKind = "PROPOSAL_OPENED" + EventProposalMerged ProposalEventKind = "PROPOSAL_MERGED" + EventProposalRejected ProposalEventKind = "PROPOSAL_REJECTED" +) + +// EventSink receives proposal lifecycle events for out-of-band delivery +// (webhooks). It is optional: a Service with no sink emits nothing. The sink +// must not block — implementations enqueue and return. +type EventSink interface { + ProposalEvent(kind ProposalEventKind, p Proposal) +} diff --git a/service/events_test.go b/service/events_test.go new file mode 100644 index 0000000000000000000000000000000000000000..4e0bf4a29aba8baffa21a9a8f4e36c102e149df6 --- /dev/null +++ b/service/events_test.go @@ -0,0 +1,200 @@ +package service + +import ( + "context" + "testing" + + "sourcecraft.dev/bigbes/sr-ht-spec/core" +) + +// recordedEvent is one call the fake sink captured. +type recordedEvent struct { + kind ProposalEventKind + prop Proposal +} + +// fakeSink records the proposal events the service emits, so a test can assert +// the kinds and the proposals carried. It does not block, matching the EventSink +// contract. +type fakeSink struct{ events []recordedEvent } + +func (f *fakeSink) ProposalEvent(kind ProposalEventKind, p Proposal) { + f.events = append(f.events, recordedEvent{kind: kind, prop: p}) +} + +func (f *fakeSink) kinds() []ProposalEventKind { + out := make([]ProposalEventKind, len(f.events)) + for i, e := range f.events { + out[i] = e.kind + } + return out +} + +// TestProposeEmitsOpened proves a newly opened proposal (no auto_merge policy) +// fires exactly one PROPOSAL_OPENED carrying that proposal, open. +func TestProposeEmitsOpened(t *testing.T) { + svc, _ := newTestService(t) + sink := &fakeSink{} + svc.SetEventSink(sink) + ctx := context.Background() + + sp, err := svc.CreateSpace(ctx, fxSpace) + if err != nil { + t.Fatalf("CreateSpace: %v", err) + } + res := openProposalFor(t, svc, sp, "specs/a.md", "S-1", mdDoc("S-1", "A", "body")) + if res.Merged { + t.Fatalf("specs/ proposal auto-merged without a policy") + } + + if got := sink.kinds(); len(got) != 1 || got[0] != EventProposalOpened { + t.Fatalf("events = %v, want exactly [PROPOSAL_OPENED]", got) + } + ev := sink.events[0] + if ev.prop.ID != res.Proposal.ID { + t.Errorf("opened proposal id = %d, want %d", ev.prop.ID, res.Proposal.ID) + } + if ev.prop.State != core.StateOpen { + t.Errorf("opened proposal state = %s, want open", ev.prop.State) + } +} + +// TestProposeAddDoesNotEmitOpened proves adding to an existing proposal is not +// a fresh PROPOSAL_OPENED — only the first write announces the proposal. +func TestProposeAddDoesNotEmitOpened(t *testing.T) { + svc, _ := newTestService(t) + sink := &fakeSink{} + svc.SetEventSink(sink) + ctx := context.Background() + + sp, err := svc.CreateSpace(ctx, fxSpace) + if err != nil { + t.Fatalf("CreateSpace: %v", err) + } + first := openProposalFor(t, svc, sp, "specs/a.md", "S-1", mdDoc("S-1", "A", "one")) + + if _, err := svc.Propose(ctx, ProposeRequest{ + Space: fxSpace, + Principal: agentPrincipal(), + ProposalID: first.Proposal.ID, + IfMatch: first.Proposal.BaseRev, + Message: "add specs/b.md", + Writes: []DocumentWrite{{Path: "specs/b.md", Content: mdDoc("S-2", "B", "two")}}, + }); err != nil { + t.Fatalf("Propose add: %v", err) + } + + if got := sink.kinds(); len(got) != 1 || got[0] != EventProposalOpened { + t.Fatalf("events = %v, want the single open from the first write only", got) + } +} + +// TestMergeEmitsMerged proves a human merge fires PROPOSAL_MERGED carrying the +// merged proposal. +func TestMergeEmitsMerged(t *testing.T) { + svc, _ := newTestService(t) + sink := &fakeSink{} + svc.SetEventSink(sink) + ctx := context.Background() + + sp, err := svc.CreateSpace(ctx, fxSpace) + if err != nil { + t.Fatalf("CreateSpace: %v", err) + } + res := openProposalFor(t, svc, sp, "specs/a.md", "S-1", mdDoc("S-1", "A", "body")) + if _, err := svc.Merge(ctx, fxSpace, res.Proposal.ID, core.ApprovalHuman); err != nil { + t.Fatalf("Merge: %v", err) + } + + if got := sink.kinds(); len(got) != 2 || + got[0] != EventProposalOpened || got[1] != EventProposalMerged { + t.Fatalf("events = %v, want [PROPOSAL_OPENED PROPOSAL_MERGED]", got) + } + merged := sink.events[1].prop + if merged.ID != res.Proposal.ID { + t.Errorf("merged proposal id = %d, want %d", merged.ID, res.Proposal.ID) + } + if merged.State != core.StateMerged { + t.Errorf("merged proposal state = %s, want merged", merged.State) + } +} + +// TestRejectEmitsRejected proves rejecting a proposal fires PROPOSAL_REJECTED +// carrying the rejected proposal. +func TestRejectEmitsRejected(t *testing.T) { + svc, _ := newTestService(t) + sink := &fakeSink{} + svc.SetEventSink(sink) + ctx := context.Background() + + sp, err := svc.CreateSpace(ctx, fxSpace) + if err != nil { + t.Fatalf("CreateSpace: %v", err) + } + res := openProposalFor(t, svc, sp, "specs/a.md", "S-1", mdDoc("S-1", "A", "body")) + if _, err := svc.Reject(ctx, fxSpace, res.Proposal.ID); err != nil { + t.Fatalf("Reject: %v", err) + } + + if got := sink.kinds(); len(got) != 2 || + got[0] != EventProposalOpened || got[1] != EventProposalRejected { + t.Fatalf("events = %v, want [PROPOSAL_OPENED PROPOSAL_REJECTED]", got) + } + rejected := sink.events[1].prop + if rejected.ID != res.Proposal.ID { + t.Errorf("rejected proposal id = %d, want %d", rejected.ID, res.Proposal.ID) + } + if rejected.State != core.StateRejected { + t.Errorf("rejected proposal state = %s, want rejected", rejected.State) + } +} + +// TestAutoMergeEmitsOpenedThenMerged proves an auto-merged proposal fires +// PROPOSAL_OPENED then PROPOSAL_MERGED, in that order, with the merge recorded +// as policy-approved. +func TestAutoMergeEmitsOpenedThenMerged(t *testing.T) { + svc, _ := newTestService(t) + sink := &fakeSink{} + svc.SetEventSink(sink) + ctx := context.Background() + + sp, err := svc.CreateSpace(ctx, fxSpace) + if err != nil { + t.Fatalf("CreateSpace: %v", err) + } + // Auto-merge everything under notes/. + commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + ".spec.yml": []byte("review:\n auto_merge: [notes/**]\n"), + }) + base, err := sp.Repo.ApprovedHead(ctx) + if err != nil { + t.Fatalf("ApprovedHead: %v", err) + } + + res, err := svc.Propose(ctx, ProposeRequest{ + Space: fxSpace, + Principal: agentPrincipal(), + Title: "firehose note", + IfMatch: base.String(), + Message: "add notes/a.md", + Writes: []DocumentWrite{{Path: "notes/a.md", Content: mdDoc("N-1", "A", "body")}}, + }) + if err != nil { + t.Fatalf("Propose: %v", err) + } + if !res.Merged { + t.Fatalf("Merged = false, want the policy to have landed it") + } + + if got := sink.kinds(); len(got) != 2 || + got[0] != EventProposalOpened || got[1] != EventProposalMerged { + t.Fatalf("events = %v, want [PROPOSAL_OPENED PROPOSAL_MERGED]", got) + } + if opened := sink.events[0].prop; opened.State != core.StateOpen { + t.Errorf("opened event state = %s, want open (before the merge landed)", opened.State) + } + merged := sink.events[1].prop + if merged.State != core.StateMerged || merged.Approval != core.ApprovalPolicy { + t.Errorf("merged event = state %s approval %s, want merged/policy", merged.State, merged.Approval) + } +} diff --git a/service/merge.go b/service/merge.go index a39488e31ba38e4bc8ff4fd770a1eea0f83872d8..51e9b7bf7104b53e748cf58602b7bdb8f8131105 100644 --- a/service/merge.go +++ b/service/merge.go @@ -45,7 +45,12 @@ func (s *Service) Reject(ctx context.Context, ref core.SpaceRef, proposalID int) if err := s.store.RejectProposal(ctx, row.ID); err != nil { return Proposal{}, resolveProposalErr(err, row.ID) } - return s.GetProposal(ctx, row.ID) + rejected, err := s.GetProposal(ctx, row.ID) + if err != nil { + return Proposal{}, err + } + s.emit(EventProposalRejected, rejected) + return rejected, nil } // mergeProposal is the merge itself, shared by the public Merge and by @@ -142,7 +147,14 @@ func (s *Service) mergeProposal(ctx context.Context, sp *Space, row *db.Proposal "(the reconciler will repair it): %w", row.ID, short(mergedRev), sp.Ref, err) } - return s.GetProposal(ctx, row.ID) + // mergeProposal is the single merge point — both the public Merge and + // auto-merge reach it — so PROPOSAL_MERGED fires here, once per merge. + merged, err := s.GetProposal(ctx, row.ID) + if err != nil { + return Proposal{}, err + } + s.emit(EventProposalMerged, merged) + return merged, nil } // openProposalRow resolves a proposal by id within a named space, opening the diff --git a/service/propose.go b/service/propose.go index 67b2613b320f4749c6c30d9721639d37c2e2daa2..5596b58349e934ca008533e3366f1a69998a5da2 100644 --- a/service/propose.go +++ b/service/propose.go @@ -130,6 +130,14 @@ func (s *Service) Propose(ctx context.Context, req ProposeRequest) (ProposeResul return ProposeResult{}, err } + // A newly opened proposal fires PROPOSAL_OPENED; adding to an existing one + // does not — the add is a revision of a proposal already announced. The event + // goes out before the auto-merge attempt so an auto-merged proposal reports + // Opened then Merged, in that order. + if req.ProposalID == 0 { + s.emit(EventProposalOpened, proposalView(row, sp.Ref)) + } + // Auto-merge: land immediately when every path the proposal changes may skip // human review under the policy at the approved head. A stale or otherwise // unlandable auto-merge leaves the proposal open — see the method doc. diff --git a/service/service.go b/service/service.go index 38a75608dd9b60840007d476abf2096abd94430b..d7fda935fec9e3281057456646e67b48635e2aae 100644 --- a/service/service.go +++ b/service/service.go @@ -224,6 +224,10 @@ type Service struct { // UserID the coreauth bridge stamps on the owner's AuthContext. ownerUserID int + // events is the optional webhook/notification sink. Nil until SetEventSink + // installs it at startup; a Service with no sink emits nothing. + events EventSink + // grace is how long a proposal row with no branch is left alone before the // reconciler deletes it. See DefaultReconcileGrace. grace time.Duration @@ -264,6 +268,17 @@ func New(cfg Config, q db.Querier) (*Service, error) { }, nil } +// SetEventSink installs the webhook/notification sink. Called once at startup, +// after the sink (which needs the owner user id) is built. +func (s *Service) SetEventSink(sink EventSink) { s.events = sink } + +// emit fires a proposal event when a sink is installed. Nil-safe. +func (s *Service) emit(kind ProposalEventKind, p Proposal) { + if s.events != nil { + s.events.ProposalEvent(kind, p) + } +} + // Config returns the configuration this service was built from. func (s *Service) Config() Config { return s.cfg }