9 files changed, 366 insertions(+), 3 deletions(-)

M cmd/specsrht/main.go
A cmd/specsrht/webhooks.go
M go.mod
A graph/webhook_event.go
A service/events.go
A service/events_test.go
M service/merge.go
M service/propose.go
M service/service.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()
 

@@ 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) }()
+}

@@ 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

@@ 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
+}

@@ 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)
+}

@@ 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)
+	}
+}

@@ 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

@@ 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.

@@ 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 }