feat(contributions): host-side contribution slots primitive

db/pool_cache.go

@@ -103,7 +103,7 @@ func createPool(ctx context.Context, cred Credential) (*pgxpool.Pool, error) {
 // might skip resetScope (caller panic, missing defer, future bug). If the
 // reset fails, the connection is destroyed instead of being reused.
 //
-// set_config(_, '', false) is used instead of RESET ms.app_id because RESET
+// set_config(_, ”, false) is used instead of RESET ms.app_id because RESET
 // errors out if the custom GUC was never set on this connection (fresh conn
 // returning to the pool for the first time).
 func afterReleaseReset(conn *pgx.Conn) bool {

internal/contributions/handlers.go

@@ -0,0 +1,148 @@
+package contributions
+
+import (
+	"errors"
+	"io"
+	"net/http"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/jackc/pgx/v5"
+
+	"github.com/mirrorstack-ai/app-module-sdk/db"
+	"github.com/mirrorstack-ai/app-module-sdk/internal/httputil"
+)
+
+// MaxPayloadBytes caps the body size for register requests. Slot
+// payloads are config metadata (icon, label, login_path) — small. A
+// 16 KB cap keeps a misbehaving contributor from blowing up the
+// host's DB row size while leaving headroom for moderately rich
+// shapes (e.g. arrays of column definitions for users-table-columns).
+const MaxPayloadBytes = 16 * 1024
+
+// DBFunc supplies a database connection scoped to the request. The
+// SDK already exposes ms.DB; the handlers take it as an interface so
+// tests can inject a fake.
+type DBFunc func(r *http.Request) (db.Querier, func(), error)
+
+// Handlers wires the registry + storage + DB accessor into chi
+// handlers the SDK can mount under /__mirrorstack/contrib.
+type Handlers struct {
+	registry *Registry
+	storage  *Storage
+	openDB   DBFunc
+}
+
+// NewHandlers constructs a handler set.
+func NewHandlers(registry *Registry, storage *Storage, openDB DBFunc) *Handlers {
+	return &Handlers{registry: registry, storage: storage, openDB: openDB}
+}
+
+// Routes returns a chi router scoped under /__mirrorstack/contrib.
+// Caller is responsible for applying auth middleware (Internal scope
+// is the expected wrapper — contributions move trusted module-to-
+// module so the secret gate guards the write path).
+func (h *Handlers) Routes() chi.Router {
+	r := chi.NewRouter()
+	r.Post("/{slot}/{id}", h.register)
+	r.Delete("/{slot}/{id}", h.unregister)
+	r.Get("/{slot}", h.list)
+	return r
+}
+
+// register handles POST /{slot}/{id}. Body is the typed payload.
+// Validated against the slot's declared T (unmarshal probe) before
+// hitting storage.
+func (h *Handlers) register(w http.ResponseWriter, r *http.Request) {
+	slotKey := chi.URLParam(r, "slot")
+	id := chi.URLParam(r, "id")
+
+	slot, ok := h.registry.Get(slotKey)
+	if !ok {
+		httputil.JSON(w, http.StatusNotFound, httputil.ErrorResponse{Error: "unknown contribution slot"})
+		return
+	}
+	if id == "" {
+		httputil.JSON(w, http.StatusBadRequest, httputil.ErrorResponse{Error: "contribution id is required"})
+		return
+	}
+
+	// Outer httputil.MaxBytes middleware (mounted by core.Module on
+	// the /__mirrorstack/contrib subtree) already caps the body —
+	// just read it.
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		httputil.JSON(w, http.StatusBadRequest, httputil.ErrorResponse{Error: "payload unreadable"})
+		return
+	}
+	if err := slot.Validate(body); err != nil {
+		httputil.JSON(w, http.StatusBadRequest, httputil.ErrorResponse{Error: "invalid payload: " + err.Error()})
+		return
+	}
+
+	q, release, err := h.openDB(r)
+	if err != nil {
+		httputil.JSON(w, http.StatusInternalServerError, httputil.ErrorResponse{Error: err.Error()})
+		return
+	}
+	defer release()
+
+	if err := h.storage.Upsert(r.Context(), q, slotKey, id, body); err != nil {
+		httputil.JSON(w, http.StatusInternalServerError, httputil.ErrorResponse{Error: err.Error()})
+		return
+	}
+	httputil.JSON(w, http.StatusOK, map[string]string{"id": id, "slot": slotKey})
+}
+
+// unregister handles DELETE /{slot}/{id}. Returns 204 on success,
+// 404 if the contribution wasn't registered.
+func (h *Handlers) unregister(w http.ResponseWriter, r *http.Request) {
+	slotKey := chi.URLParam(r, "slot")
+	id := chi.URLParam(r, "id")
+
+	if _, ok := h.registry.Get(slotKey); !ok {
+		httputil.JSON(w, http.StatusNotFound, httputil.ErrorResponse{Error: "unknown contribution slot"})
+		return
+	}
+
+	q, release, err := h.openDB(r)
+	if err != nil {
+		httputil.JSON(w, http.StatusInternalServerError, httputil.ErrorResponse{Error: err.Error()})
+		return
+	}
+	defer release()
+
+	if err := h.storage.Delete(r.Context(), q, slotKey, id); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			httputil.JSON(w, http.StatusNotFound, httputil.ErrorResponse{Error: "contribution not found"})
+			return
+		}
+		httputil.JSON(w, http.StatusInternalServerError, httputil.ErrorResponse{Error: err.Error()})
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+// list handles GET /{slot}. Returns the registered contributions
+// sorted newest-first.
+func (h *Handlers) list(w http.ResponseWriter, r *http.Request) {
+	slotKey := chi.URLParam(r, "slot")
+
+	if _, ok := h.registry.Get(slotKey); !ok {
+		httputil.JSON(w, http.StatusNotFound, httputil.ErrorResponse{Error: "unknown contribution slot"})
+		return
+	}
+
+	q, release, err := h.openDB(r)
+	if err != nil {
+		httputil.JSON(w, http.StatusInternalServerError, httputil.ErrorResponse{Error: err.Error()})
+		return
+	}
+	defer release()
+
+	out, err := h.storage.List(r.Context(), q, slotKey)
+	if err != nil {
+		httputil.JSON(w, http.StatusInternalServerError, httputil.ErrorResponse{Error: err.Error()})
+		return
+	}
+	httputil.JSON(w, http.StatusOK, map[string]any{"contributions": out})
+}

internal/contributions/storage.go

@@ -0,0 +1,132 @@
+package contributions
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+
+	"github.com/mirrorstack-ai/app-module-sdk/db"
+)
+
+// listLimit caps the row count returned by Storage.List so a slot
+// with thousands of contributions (unlikely today) can't return a
+// pathologically large response. Picked to comfortably cover every
+// realistic slot count in v1.
+const listLimit = 1000
+
+// Storage wraps the contributions table CRUD. Sanitized name is
+// computed once at construction so the SQL builders don't pay the
+// identifier-escape cost per call.
+type Storage struct {
+	table string // already pgx-Sanitize'd, safe to interpolate
+}
+
+// NewStorage constructs a Storage rooted at the given module ID.
+// modulePrefix is validated upstream against moduleIDPattern in
+// core.New(), so Sanitize is belt-and-suspenders.
+func NewStorage(modulePrefix string) *Storage {
+	return &Storage{table: pgx.Identifier{modulePrefix + "_contributions"}.Sanitize()}
+}
+
+// TableName returns the SQL-safe (already sanitized + quoted)
+// contributions table name. Exposed for test diagnostics.
+func (s *Storage) TableName() string { return s.table }
+
+// EnsureTable runs CREATE TABLE IF NOT EXISTS. Dev fallback; prod
+// schema management lives in the lifecycle install hook.
+func (s *Storage) EnsureTable(ctx context.Context, q db.Querier) error {
+	// Index name strips the surrounding quotes from the sanitized
+	// table name so it stays a bare identifier (Postgres errors on
+	// quoted index names that contain double-quotes).
+	idx := s.tableUnquoted() + "_slot_registered_idx"
+	stmt := fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %[1]s (
+		    slot            text NOT NULL,
+		    contribution_id text NOT NULL,
+		    payload         jsonb NOT NULL,
+		    registered_at   timestamptz NOT NULL DEFAULT now(),
+		    PRIMARY KEY (slot, contribution_id)
+		);
+
+		CREATE INDEX IF NOT EXISTS %[2]s
+		    ON %[1]s (slot, registered_at DESC);`,
+		s.table, idx,
+	)
+	_, err := q.Exec(ctx, stmt)
+	return err
+}
+
+// tableUnquoted returns the table name without the pgx-Sanitize
+// surrounding double quotes — needed when building an index name
+// (which must itself be an identifier, not a quoted-string).
+func (s *Storage) tableUnquoted() string {
+	t := s.table
+	if len(t) >= 2 && t[0] == '"' && t[len(t)-1] == '"' {
+		return t[1 : len(t)-1]
+	}
+	return t
+}
+
+// Upsert writes a contribution. Idempotent on (slot, contribution_id).
+func (s *Storage) Upsert(ctx context.Context, q db.Querier, slot, id string, payload json.RawMessage) error {
+	if id == "" {
+		return ErrEmptyID
+	}
+	if !json.Valid(payload) {
+		return errors.Join(ErrInvalidPayload, errors.New("storage: payload is not valid JSON"))
+	}
+	stmt := fmt.Sprintf(`
+		INSERT INTO %s (slot, contribution_id, payload, registered_at)
+		VALUES ($1, $2, $3, now())
+		ON CONFLICT (slot, contribution_id)
+		DO UPDATE SET payload = EXCLUDED.payload, registered_at = now()`,
+		s.table,
+	)
+	_, err := q.Exec(ctx, stmt, slot, id, payload)
+	return err
+}
+
+// Delete removes a contribution. Returns pgx.ErrNoRows when the row
+// didn't exist; the HTTP layer maps that to 404 (double-delete is
+// not silently idempotent here).
+func (s *Storage) Delete(ctx context.Context, q db.Querier, slot, id string) error {
+	stmt := fmt.Sprintf(`DELETE FROM %s WHERE slot = $1 AND contribution_id = $2`, s.table)
+	tag, err := q.Exec(ctx, stmt, slot, id)
+	if err != nil {
+		return err
+	}
+	if tag.RowsAffected() == 0 {
+		return pgx.ErrNoRows
+	}
+	return nil
+}
+
+// List returns up to listLimit contributions registered against
+// `slot`, newest first.
+func (s *Storage) List(ctx context.Context, q db.Querier, slot string) ([]Contribution, error) {
+	stmt := fmt.Sprintf(`
+		SELECT contribution_id, payload, registered_at
+		FROM %s
+		WHERE slot = $1
+		ORDER BY registered_at DESC
+		LIMIT %d`,
+		s.table, listLimit,
+	)
+	rows, err := q.Query(ctx, stmt, slot)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := []Contribution{}
+	for rows.Next() {
+		var c Contribution
+		if err := rows.Scan(&c.ID, &c.Payload, &c.RegisteredAt); err != nil {
+			return nil, err
+		}
+		out = append(out, c)
+	}
+	return out, rows.Err()
+}