Add Slack org lifecycle notifications

README.md

@@ -260,6 +260,10 @@ Run with config file:
 | `POSTHOG_HOST` | PostHog ingest host | `us.i.posthog.com` |
 | `ADDITIONAL_POSTHOG_API_KEYS` | **(Experimental)** Comma-separated list of additional PostHog API keys to publish logs to. Requires `POSTHOG_API_KEY` to be set. | - |
 | `DUCKGRES_IDENTIFIER` | Suffix appended to the OTel `service.name` in PostHog logs (e.g., `duckgres-acme`); only used when `POSTHOG_API_KEY` is set | - |
+| `DUCKGRES_SLACK_WEBHOOK_URL` | Slack incoming webhook URL for operational notifications. Env-only because it is secret-bearing. | - |
+| `DUCKGRES_NOTIFICATION_HASH_KEY` | Optional secret key used to include a stable HMAC org reference in Slack notifications. Env-only because it is secret-bearing. | - |
+| `DUCKGRES_NOTIFICATION_QUEUE_SIZE` | In-process notification queue size before events are dropped instead of blocking user requests | `100` |
+| `DUCKGRES_NOTIFICATION_TIMEOUT` | Per-notification delivery timeout | `2s` |
 
 ### PostHog Logging
 
@@ -329,6 +333,29 @@ teardown), so you can build a provisioning funnel and alert on failures.
 > scales 1:1 with query throughput. Capture is asynchronous and batched, so it
 > stays off the query latency path.
 
+### Slack Operational Notifications
+
+Set `DUCKGRES_SLACK_WEBHOOK_URL` to enable fire-and-forget Slack notifications
+for org and managed-warehouse lifecycle events. Delivery is asynchronous:
+Duckgres queues events in memory, applies `DUCKGRES_NOTIFICATION_TIMEOUT`
+(default `2s`) per webhook call, and drops notifications when the bounded queue
+is full (default size `100`) rather than blocking signup, provisioning, or
+admin requests. Shutdown does not drain the in-memory notification queue; this
+preserves the best-effort contract during process exit.
+
+Slack messages intentionally do **not** include raw org IDs, customer names,
+database names, endpoints, SQL text, credentials, secret names, or bucket
+names. They include the event name and a small allowlist of low-risk properties
+such as `metadata_store`, `ducklake_enabled`, `source`, and stable failure
+`reason`. Set `DUCKGRES_NOTIFICATION_HASH_KEY` to add a stable
+`hmac-sha256` org reference for Slack-side correlation.
+
+See [`docs/runbooks/slack-notifications.md`](docs/runbooks/slack-notifications.md)
+for local testing and failure recovery.
+
+Notification health is exposed through `duckgres_notifications_dropped_total`
+and `duckgres_notification_delivery_failures_total`.
+
 ### CLI Flags
 
 ```bash

cmd/duckgres-controlplane/main.go

@@ -145,6 +145,8 @@ func main() {
 	defer loggingShutdown()
 	analyticsShutdown := cliboot.InitAnalytics()
 	defer analyticsShutdown()
+	notificationsShutdown := cliboot.InitNotifications()
+	defer notificationsShutdown()
 	tracingShutdown := cliboot.InitTracing()
 	defer tracingShutdown()
 

controlplane/admin/api.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/posthog/duckgres/controlplane/configstore"
+	"github.com/posthog/duckgres/internal/notifications"
 	"gorm.io/gorm"
 	"gorm.io/gorm/clause"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -554,6 +555,11 @@ func (h *apiHandler) createOrg(c *gin.Context) {
 		c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
 		return
 	}
+	notifications.Default().Notify(notifications.Event{
+		Name:  "org_created",
+		OrgID: org.Name,
+		Props: map[string]any{"source": "admin_api"},
+	})
 	c.JSON(http.StatusCreated, org)
 }
 
@@ -681,6 +687,11 @@ func (h *apiHandler) deleteOrg(c *gin.Context) {
 		c.JSON(http.StatusNotFound, gin.H{"error": "org not found"})
 		return
 	}
+	notifications.Default().Notify(notifications.Event{
+		Name:  "org_deleted",
+		OrgID: name,
+		Props: map[string]any{"source": "admin_api"},
+	})
 	c.JSON(http.StatusOK, gin.H{"deleted": name})
 }
 

controlplane/admin/api_test.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/posthog/duckgres/controlplane/configstore"
+	"github.com/posthog/duckgres/internal/notifications"
 	"gorm.io/gorm"
 )
 
@@ -24,6 +25,39 @@ type fakeAPIStore struct {
 	warehouses map[string]*configstore.ManagedWarehouse
 }
 
+type adminCapturedNotification struct {
+	name  string
+	orgID string
+	props map[string]any
+}
+
+type adminFakeNotifier struct {
+	events []adminCapturedNotification
+}
+
+func (f *adminFakeNotifier) Notify(event notifications.Event) {
+	f.events = append(f.events, adminCapturedNotification{name: event.Name, orgID: event.OrgID, props: event.Props})
+}
+func (f *adminFakeNotifier) Close() {}
+
+func installAdminFakeNotifier(t *testing.T) *adminFakeNotifier {
+	t.Helper()
+	fake := &adminFakeNotifier{}
+	notifications.SetDefault(fake)
+	t.Cleanup(func() { notifications.SetDefault(nil) })
+	return fake
+}
+
+func (f *adminFakeNotifier) count(name string) int {
+	n := 0
+	for _, event := range f.events {
+		if event.name == name {
+			n++
+		}
+	}
+	return n
+}
+
 func newFakeAPIStore() *fakeAPIStore {
 	return &fakeAPIStore{
 		orgs:       make(map[string]*configstore.Org),
@@ -1139,6 +1173,7 @@ func TestDeleteOrgRejectsWhenWarehouseStillExists(t *testing.T) {
 }
 
 func TestDeleteOrgSucceedsAfterWarehouseRemoved(t *testing.T) {
+	notifier := installAdminFakeNotifier(t)
 	store := newFakeAPIStore()
 	seedOrgWithWarehouse(store, "analytics")
 	delete(store.warehouses, "analytics")
@@ -1154,9 +1189,13 @@ func TestDeleteOrgSucceedsAfterWarehouseRemoved(t *testing.T) {
 	if _, ok := store.orgs["analytics"]; ok {
 		t.Fatal("expected org to be deleted once the warehouse is gone")
 	}
+	if got := notifier.count("org_deleted"); got != 1 {
+		t.Fatalf("org_deleted notifications = %d, want 1 (all: %+v)", got, notifier.events)
+	}
 }
 
 func TestDeleteOrgCascadesDeletedWarehouse(t *testing.T) {
+	notifier := installAdminFakeNotifier(t)
 	store := newFakeAPIStore()
 	seedOrgWithWarehouse(store, "analytics")
 	// Simulate a completed deprovision: the provisioner leaves the warehouse
@@ -1177,6 +1216,9 @@ func TestDeleteOrgCascadesDeletedWarehouse(t *testing.T) {
 	if _, ok := store.warehouses["analytics"]; ok {
 		t.Fatal("expected the deleted warehouse row to be cascaded away")
 	}
+	if got := notifier.count("org_deleted"); got != 1 {
+		t.Fatalf("org_deleted notifications = %d, want 1 (all: %+v)", got, notifier.events)
+	}
 }
 
 func TestGetOrgOmitsMinWorkers(t *testing.T) {

controlplane/provisioner/controller.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/posthog/duckgres/controlplane/configstore"
 	"github.com/posthog/duckgres/internal/analytics"
+	"github.com/posthog/duckgres/internal/notifications"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 )
 
@@ -35,6 +36,7 @@ func captureProvisionFailed(w *configstore.ManagedWarehouse, reason string, upda
 	props := provisionEventProps(w)
 	props["reason"] = reason
 	analytics.Default().Capture("warehouse_provision_failed", w.OrgID, props)
+	notifications.Default().Notify(notifications.Event{Name: "warehouse_provision_failed", OrgID: w.OrgID, Props: props})
 }
 
 // ducklingCRName returns the Duckling CR name for a warehouse row: the stored
@@ -354,7 +356,9 @@ func (c *Controller) reconcileProvisioning(ctx context.Context, w *configstore.M
 			// usable. Guarded on a successful CAS from Provisioning, so this
 			// fires exactly once — the next tick routes a Ready warehouse to
 			// reconcileReady, not here.
-			analytics.Default().Capture("warehouse_provision_success", w.OrgID, provisionEventProps(w))
+			props := provisionEventProps(w)
+			analytics.Default().Capture("warehouse_provision_success", w.OrgID, props)
+			notifications.Default().Notify(notifications.Event{Name: "warehouse_provision_success", OrgID: w.OrgID, Props: props})
 		}
 	}
 
@@ -535,6 +539,11 @@ func (c *Controller) reconcileDeleting(ctx context.Context, w *configstore.Manag
 			analytics.Default().Capture("warehouse_deprovision_failed", w.OrgID, map[string]any{
 				"reason": "duckling_delete_failed",
 			})
+			notifications.Default().Notify(notifications.Event{
+				Name:  "warehouse_deprovision_failed",
+				OrgID: w.OrgID,
+				Props: map[string]any{"reason": "duckling_delete_failed"},
+			})
 			return
 		}
 	}
@@ -548,5 +557,6 @@ func (c *Controller) reconcileDeleting(ctx context.Context, w *configstore.Manag
 		// Terminal success: all underlying resources are gone. Guarded on a
 		// successful CAS from Deleting, so this fires exactly once.
 		analytics.Default().Capture("warehouse_deprovision_success", w.OrgID, nil)
+		notifications.Default().Notify(notifications.Event{Name: "warehouse_deprovision_success", OrgID: w.OrgID})
 	}
 }

controlplane/provisioner/controller_analytics_test.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/posthog/duckgres/controlplane/configstore"
 	"github.com/posthog/duckgres/internal/analytics"
+	"github.com/posthog/duckgres/internal/notifications"
 )
 
 // capturedEvent records one analytics.Capture call.
@@ -44,6 +45,53 @@ func installFakeTracker(t *testing.T) *fakeTracker {
 	return fake
 }
 
+type capturedNotification struct {
+	event string
+	orgID string
+	props map[string]any
+}
+
+type fakeNotifier struct {
+	events []capturedNotification
+}
+
+func (f *fakeNotifier) Notify(event notifications.Event) {
+	f.events = append(f.events, capturedNotification{event: event.Name, orgID: event.OrgID, props: event.Props})
+}
+func (f *fakeNotifier) Close() {}
+
+func installFakeNotifier(t *testing.T) *fakeNotifier {
+	t.Helper()
+	fake := &fakeNotifier{}
+	notifications.SetDefault(fake)
+	t.Cleanup(func() { notifications.SetDefault(nil) })
+	return fake
+}
+
+func (f *fakeNotifier) only(t *testing.T, event string) capturedNotification {
+	t.Helper()
+	var found []capturedNotification
+	for _, e := range f.events {
+		if e.event == event {
+			found = append(found, e)
+		}
+	}
+	if len(found) != 1 {
+		t.Fatalf("expected exactly one %q notification, got %d (all: %+v)", event, len(found), f.events)
+	}
+	return found[0]
+}
+
+func (f *fakeNotifier) count(event string) int {
+	n := 0
+	for _, e := range f.events {
+		if e.event == event {
+			n++
+		}
+	}
+	return n
+}
+
 // only returns the single event with the given name, failing if there isn't
 // exactly one.
 func (f *fakeTracker) only(t *testing.T, event string) capturedEvent {
@@ -76,6 +124,7 @@ func (f *fakeTracker) count(event string) int {
 // warehouse metadata, and no failure event.
 func TestReconcileProvisioningSuccessEmitsEvent(t *testing.T) {
 	fake := installFakeTracker(t)
+	notifier := installFakeNotifier(t)
 	dc, fakeK8s := newFakeDucklingClient()
 	fs := newFakeStore()
 	fs.warehouses["org-ok"] = &configstore.ManagedWarehouse{
@@ -132,19 +181,30 @@ func TestReconcileProvisioningSuccessEmitsEvent(t *testing.T) {
 	if n := fake.count("warehouse_provision_failed"); n != 0 {
 		t.Errorf("expected no failure event, got %d", n)
 	}
+	notification := notifier.only(t, "warehouse_provision_success")
+	if notification.orgID != "org-ok" {
+		t.Errorf("notification orgID = %q, want org-ok", notification.orgID)
+	}
+	if notification.props["metadata_store"] != string(configstore.MetadataStoreKindExternal) {
+		t.Errorf("notification metadata_store = %v, want external", notification.props["metadata_store"])
+	}
 
 	// A second reconcile finds the warehouse already Ready (routed to
 	// reconcileReady), so the success event must not fire again.
 	ctrl.reconcile(ctx)
 	if n := fake.count("warehouse_provision_success"); n != 1 {
 		t.Errorf("expected exactly one success event across two reconciles, got %d", n)
 	}
+	if n := notifier.count("warehouse_provision_success"); n != 1 {
+		t.Errorf("expected exactly one success notification across two reconciles, got %d", n)
+	}
 }
 
 // TestReconcileProvisioningTimeoutEmitsFailure asserts the 30-minute timeout
 // path emits warehouse_provision_failed with reason=provisioning_timeout.
 func TestReconcileProvisioningTimeoutEmitsFailure(t *testing.T) {
 	fake := installFakeTracker(t)
+	notifier := installFakeNotifier(t)
 	dc, _ := newFakeDucklingClient()
 	fs := newFakeStore()
 	fs.warehouses["org-slow"] = &configstore.ManagedWarehouse{
@@ -171,13 +231,18 @@ func TestReconcileProvisioningTimeoutEmitsFailure(t *testing.T) {
 	if e.props["metadata_store"] != string(configstore.MetadataStoreKindCnpgShard) {
 		t.Errorf("metadata_store = %v, want cnpg-shard", e.props["metadata_store"])
 	}
+	notification := notifier.only(t, "warehouse_provision_failed")
+	if notification.props["reason"] != "provisioning_timeout" {
+		t.Errorf("notification reason = %v, want provisioning_timeout", notification.props["reason"])
+	}
 }
 
 // TestReconcileProvisioningCrossplaneFailureEmitsFailure asserts that a
 // persistent Crossplane sync failure (>10 min) emits warehouse_provision_failed
 // with reason=crossplane_sync_failure.
 func TestReconcileProvisioningCrossplaneFailureEmitsFailure(t *testing.T) {
 	fake := installFakeTracker(t)
+	notifier := installFakeNotifier(t)
 	dc, fakeK8s := newFakeDucklingClient()
 	fs := newFakeStore()
 	fs.warehouses["org-xp"] = &configstore.ManagedWarehouse{
@@ -216,12 +281,17 @@ func TestReconcileProvisioningCrossplaneFailureEmitsFailure(t *testing.T) {
 	if n := fake.count("warehouse_provision_success"); n != 0 {
 		t.Errorf("expected no success event, got %d", n)
 	}
+	notification := notifier.only(t, "warehouse_provision_failed")
+	if notification.props["reason"] != "crossplane_sync_failure" {
+		t.Errorf("notification reason = %v, want crossplane_sync_failure", notification.props["reason"])
+	}
 }
 
 // TestReconcileDeletingSuccessEmitsEvent asserts that a completed teardown
 // emits exactly one warehouse_deprovision_success and no failure event.
 func TestReconcileDeletingSuccessEmitsEvent(t *testing.T) {
 	fake := installFakeTracker(t)
+	notifier := installFakeNotifier(t)
 	dc, fakeK8s := newFakeDucklingClient()
 	fs := newFakeStore()
 	fs.warehouses["org-del"] = &configstore.ManagedWarehouse{
@@ -251,13 +321,17 @@ func TestReconcileDeletingSuccessEmitsEvent(t *testing.T) {
 	if n := fake.count("warehouse_deprovision_failed"); n != 0 {
 		t.Errorf("expected no failure event, got %d", n)
 	}
+	if e := notifier.only(t, "warehouse_deprovision_success"); e.orgID != "org-del" {
+		t.Errorf("notification orgID = %q, want org-del", e.orgID)
+	}
 }
 
 // TestReconcileDeletingFailureEmitsEvent asserts that a non-NotFound error
 // deleting the Duckling CR emits warehouse_deprovision_failed and leaves the
 // warehouse in Deleting (so the controller retries).
 func TestReconcileDeletingFailureEmitsEvent(t *testing.T) {
 	fake := installFakeTracker(t)
+	notifier := installFakeNotifier(t)
 	dc, fakeK8s := newFakeDucklingClient()
 	fs := newFakeStore()
 	fs.warehouses["org-stuck"] = &configstore.ManagedWarehouse{
@@ -297,4 +371,8 @@ func TestReconcileDeletingFailureEmitsEvent(t *testing.T) {
 	if n := fake.count("warehouse_deprovision_success"); n != 0 {
 		t.Errorf("expected no success event, got %d", n)
 	}
+	notification := notifier.only(t, "warehouse_deprovision_failed")
+	if notification.props["reason"] != "duckling_delete_failed" {
+		t.Errorf("notification reason = %v, want duckling_delete_failed", notification.props["reason"])
+	}
 }