ProofedMesh/services/dummy-sink/internal/httpapi/server.go

// Package httpapi exposes dummy-sink's single endpoint:
// POST /api/v1/sink/dummy/deliver per the DEV-611 contract.
//
// Flow:
//  1. Parse the request body ({decision_id, payload_hash, payload,
//     approval_ref, transform_ref}).
//  2. Call verification-service /api/v1/verify/chain.
//  3. On valid=true:
//     a. Write payload atomically to <DataDir>/<decision_id>.json.
//     b. Build the handover_attestation (schema
//        basicstack.de/attestations/handover/v1).
//     c. Sign it via signing-service.
//     d. Submit the DSSE envelope to the transparency-log.
//     e. Reply 201 Created {handover_ref, log_index}.
//  4. On valid=false:
//     a. If verification-service flagged `replay: true`, reply 409.
//     b. Otherwise reply 403 Forbidden {reasons: [...]}.
//
// Replay protection intentionally lives in the verification-service per
// the parent plan: dummy-sink keeps no replay table of its own.
package httpapi

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"strings"
	"time"

	"basicstack.de/proofedmesh/services/dummy-sink/internal/canonical"
	"basicstack.de/proofedmesh/services/dummy-sink/internal/dsse"
	"basicstack.de/proofedmesh/services/dummy-sink/internal/handover"
	"basicstack.de/proofedmesh/services/dummy-sink/internal/signer"
	"basicstack.de/proofedmesh/services/dummy-sink/internal/sink"
	"basicstack.de/proofedmesh/services/dummy-sink/internal/tlog"
	"basicstack.de/proofedmesh/services/dummy-sink/internal/verify"
)

// Verifier calls the verification-service.
type Verifier interface {
	Verify(ctx context.Context, req verify.Request) (verify.Result, error)
}

// Signer calls the signing-service.
type Signer interface {
	Sign(ctx context.Context, keyRef string, stmt signer.Statement) (signer.SignedResult, error)
}

// TLog submits DSSE envelopes to the transparency-log.
type TLog interface {
	Submit(ctx context.Context, env dsse.Envelope, contentHash string) (tlog.Receipt, error)
}

// Store writes payloads to disk atomically.
type Store interface {
	Write(decisionID string, payload any) (sink.WriteResult, error)
}

// Server wires the delivery pipeline.
type Server struct {
	Verifier      Verifier
	Store         Store
	Signer        Signer
	TLog          TLog
	SinkID        string // e.g. "sink:dummy-sink"
	Actor         string // agent registry id of the dummy-sink identity
	SigningKeyRef string // e.g. "trust-api/agent-dummy-sink-2026-q4"

	Logger *slog.Logger
	Now    func() time.Time
}

// Mount registers the endpoints on mux.
func (s Server) Mount(mux *http.ServeMux) {
	mux.HandleFunc("POST /api/v1/sink/dummy/deliver", s.handleDeliver)
	mux.HandleFunc("GET /healthz", s.handleHealth)
}

type deliverRequest struct {
	DecisionID   string          `json:"decision_id"`
	PayloadHash  string          `json:"payload_hash"`
	Payload      json.RawMessage `json:"payload"`
	ApprovalRef  string          `json:"approval_ref"`
	TransformRef string          `json:"transform_ref"`
}

type deliverResponse struct {
	HandoverRef string `json:"handover_ref"`
	LogIndex    int64  `json:"log_index"`
	DecisionID  string `json:"decision_id"`
}

type errResponse struct {
	Error errBody `json:"error"`
}

type errBody struct {
	Code    string   `json:"code"`
	Message string   `json:"message"`
	Reasons []string `json:"reasons,omitempty"`
}

func (s Server) handleHealth(w http.ResponseWriter, _ *http.Request) {
	w.WriteHeader(http.StatusOK)
	_, _ = w.Write([]byte(`{"status":"ok"}`))
}

func (s Server) handleDeliver(w http.ResponseWriter, r *http.Request) {
	logger := s.logger()
	ctx := r.Context()

	raw, err := io.ReadAll(io.LimitReader(r.Body, 10<<20))
	if err != nil {
		writeErr(w, http.StatusBadRequest, "invalid_body", "read body: "+err.Error(), nil)
		return
	}
	defer r.Body.Close()

	var req deliverRequest
	if err := json.Unmarshal(raw, &req); err != nil {
		writeErr(w, http.StatusBadRequest, "invalid_body", "parse body: "+err.Error(), nil)
		return
	}
	if err := validateRequest(req); err != nil {
		writeErr(w, http.StatusBadRequest, "invalid_body", err.Error(), nil)
		return
	}

	payloadRaw := []byte(req.Payload)
	rawDigest := sha256.Sum256(payloadRaw)
	rawHex := "sha256:" + hex.EncodeToString(rawDigest[:])

	var payloadAny any
	if err := json.Unmarshal(payloadRaw, &payloadAny); err != nil {
		writeErr(w, http.StatusBadRequest, "invalid_body", "payload must be valid JSON: "+err.Error(), nil)
		return
	}
	canonicalHash, _, err := canonical.Hash(payloadAny)
	if err != nil {
		writeErr(w, http.StatusBadRequest, "invalid_body", "canonicalise payload: "+err.Error(), nil)
		return
	}
	if req.PayloadHash != rawHex && req.PayloadHash != canonicalHash {
		writeErr(w, http.StatusBadRequest, "payload_hash_mismatch",
			fmt.Sprintf("payload_hash %q matches neither raw (%s) nor canonical (%s)",
				req.PayloadHash, rawHex, canonicalHash), nil)
		return
	}

	verdict, err := s.Verifier.Verify(ctx, verify.Request{
		DecisionID:   req.DecisionID,
		PayloadHash:  req.PayloadHash,
		ApprovalRef:  req.ApprovalRef,
		TransformRef: req.TransformRef,
		SinkTarget:   s.SinkID,
	})
	if err != nil {
		logger.Error("verify.chain failed", "decision_id", req.DecisionID, "err", err)
		if errors.Is(err, verify.ErrVerifyUnavailable) {
			writeErr(w, http.StatusBadGateway, "verify_unavailable", err.Error(), nil)
			return
		}
		writeErr(w, http.StatusBadGateway, "verify_failed", err.Error(), nil)
		return
	}

	if !verdict.Valid {
		if verdict.Replay {
			writeErr(w, http.StatusConflict, "replay", "decision_id already consumed", verdict.Reasons)
			return
		}
		writeErr(w, http.StatusForbidden, "chain_invalid", "verification chain rejected delivery", verdict.Reasons)
		return
	}

	write, err := s.Store.Write(req.DecisionID, payloadAny)
	if err != nil {
		logger.Error("store.Write failed", "decision_id", req.DecisionID, "err", err)
		if errors.Is(err, sink.ErrInvalidDecisionID) {
			writeErr(w, http.StatusBadRequest, "invalid_decision_id", err.Error(), nil)
			return
		}
		writeErr(w, http.StatusInternalServerError, "sink_write_failed", err.Error(), nil)
		return
	}

	stmt, err := handover.Build(handover.Input{
		DecisionID:    req.DecisionID,
		PayloadHash:   req.PayloadHash,
		ApprovalRef:   req.ApprovalRef,
		TransformRef:  req.TransformRef,
		SinkID:        s.SinkID,
		SinkRecordRef: "file://" + write.Path,
		Actor:         s.Actor,
		IssuedAt:      s.now(),
	})
	if err != nil {
		logger.Error("handover.Build failed", "decision_id", req.DecisionID, "err", err)
		writeErr(w, http.StatusInternalServerError, "handover_build_failed", err.Error(), nil)
		return
	}

	signed, err := s.Signer.Sign(ctx, s.SigningKeyRef, stmt)
	if err != nil {
		logger.Error("signer.Sign failed", "decision_id", req.DecisionID, "err", err)
		if errors.Is(err, signer.ErrSigningDenied) {
			writeErr(w, http.StatusForbidden, "signing_denied", err.Error(), nil)
			return
		}
		if errors.Is(err, signer.ErrSignerUnavailable) {
			writeErr(w, http.StatusBadGateway, "signer_unavailable", err.Error(), nil)
			return
		}
		writeErr(w, http.StatusInternalServerError, "sign_failed", err.Error(), nil)
		return
	}

	receipt, err := s.TLog.Submit(ctx, signed.Envelope, signed.ContentHash)
	if err != nil {
		logger.Error("tlog.Submit failed", "decision_id", req.DecisionID, "err", err)
		writeErr(w, http.StatusBadGateway, "tlog_failed", err.Error(), nil)
		return
	}

	writeJSON(w, http.StatusCreated, deliverResponse{
		HandoverRef: signed.ContentHash,
		LogIndex:    receipt.LogIndex,
		DecisionID:  req.DecisionID,
	})
}

func validateRequest(req deliverRequest) error {
	if strings.TrimSpace(req.DecisionID) == "" {
		return errors.New("decision_id is required")
	}
	if strings.TrimSpace(req.PayloadHash) == "" {
		return errors.New("payload_hash is required")
	}
	if !strings.HasPrefix(req.PayloadHash, "sha256:") || len(req.PayloadHash) != 7+64 {
		return errors.New("payload_hash must be sha256:<64-hex>")
	}
	if len(req.Payload) == 0 {
		return errors.New("payload is required")
	}
	if strings.TrimSpace(req.ApprovalRef) == "" {
		return errors.New("approval_ref is required")
	}
	if strings.TrimSpace(req.TransformRef) == "" {
		return errors.New("transform_ref is required")
	}
	return nil
}

func writeErr(w http.ResponseWriter, status int, code, msg string, reasons []string) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(errResponse{Error: errBody{
		Code: code, Message: msg, Reasons: reasons,
	}})
}

func writeJSON(w http.ResponseWriter, status int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(v)
}

func (s Server) logger() *slog.Logger {
	if s.Logger != nil {
		return s.Logger
	}
	return slog.Default()
}

func (s Server) now() time.Time {
	if s.Now != nil {
		return s.Now()
	}
	return time.Now()
}