feat: add notification delivery foundation #9

Merged
m merged 1 commits from codex/prospective-notification-delivery-v1 into main 2026-08-16 16:08:47 +00:00
18 changed files with 1184 additions and 24 deletions
+194
View File
@@ -0,0 +1,194 @@
package notificationexecutor
import (
"bytes"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"strings"
"git.cin.su/m/Assistant/contracts/capability"
commitcontract "git.cin.su/m/Assistant/contracts/commit"
capabilitydomain "git.cin.su/m/Assistant/modules/capability/domain"
"git.cin.su/m/Assistant/modules/commit/ports"
)
const maxResponseBytes = 16 << 10
var ErrInvalidNotification = errors.New("invalid prospective notification")
type Payload struct {
ActionIntentID commitcontract.ActionIntentID `json:"action_intent_id"`
TemplateCode string `json:"template_code"`
ProspectiveItemID string `json:"prospective_item_id"`
ExpectedRevision int64 `json:"expected_revision"`
Title string `json:"title"`
Message string `json:"message"`
}
type DeliveryRequest struct {
Payload
RequestDigest string `json:"request_digest"`
}
type Receipt struct {
ReceiptID string `json:"receipt_id"`
ActionIntentID commitcontract.ActionIntentID `json:"action_intent_id"`
RequestDigest string `json:"request_digest"`
AcceptedAt int64 `json:"accepted_at_unix_millis"`
}
func Digest(payload Payload) (string, error) {
if payload.ActionIntentID == "" || payload.TemplateCode != commitcontract.ProspectiveDeliveredTemplateV1 ||
payload.ProspectiveItemID == "" || payload.ExpectedRevision < 1 || strings.TrimSpace(payload.Title) == "" ||
strings.TrimSpace(payload.Message) == "" || len(payload.Title) > 160 || len(payload.Message) > 2048 {
return "", ErrInvalidNotification
}
encoded, err := json.Marshal(payload)
if err != nil {
return "", err
}
sum := sha256.Sum256(encoded)
return hex.EncodeToString(sum[:]), nil
}
func PayloadFromAction(action commitcontract.AuthorizedAction) (Payload, error) {
subject := action.Proposal.NotificationSubject
if action.Proposal.CapabilityID != commitcontract.CapabilityProspectiveNotification ||
action.Proposal.Operation != commitcontract.OperationDeliverProspectiveNotification ||
subject.BindingID == "" || subject.BindingRevision < 1 {
return Payload{}, ErrInvalidNotification
}
payload := Payload{ActionIntentID: action.Proposal.IntentID, TemplateCode: subject.TemplateCode,
ProspectiveItemID: string(subject.ProspectiveItemID), ExpectedRevision: int64(subject.ExpectedRevision),
Title: subject.RenderedTitle, Message: subject.RenderedMessage}
digest, err := Digest(payload)
if err != nil || digest != subject.RequestDigest {
return Payload{}, ErrInvalidNotification
}
return payload, nil
}
type Executor struct {
binding capability.Binding
secret string
client *http.Client
}
func New(binding capability.Binding, secret string, client *http.Client) (*Executor, error) {
if err := capabilitydomain.UsableBinding(binding); err != nil || strings.TrimSpace(secret) == "" || client == nil {
return nil, ErrInvalidNotification
}
copyClient := *client
copyClient.CheckRedirect = func(_ *http.Request, _ []*http.Request) error { return http.ErrUseLastResponse }
return &Executor{binding: binding, secret: secret, client: &copyClient}, nil
}
func (e *Executor) Execute(action commitcontract.AuthorizedAction) ports.ExecutionResult {
payload, err := e.requestFor(action)
if err != nil {
return ports.ExecutionResult{Known: true}
}
body, err := json.Marshal(DeliveryRequest{Payload: payload, RequestDigest: action.Proposal.NotificationSubject.RequestDigest})
if err != nil {
return ports.ExecutionResult{Known: true}
}
endpoint, err := e.endpoint(e.binding.DeliveryPath, string(action.Proposal.IntentID))
if err != nil {
return ports.ExecutionResult{Known: true}
}
request, err := http.NewRequest(http.MethodPut, endpoint, bytes.NewReader(body))
if err != nil {
return ports.ExecutionResult{Known: true}
}
request.Header.Set("Authorization", "Bearer "+e.secret)
request.Header.Set("Content-Type", "application/json")
request.Header.Set("Accept", "application/json")
response, err := e.client.Do(request)
if err != nil {
return ports.ExecutionResult{}
}
defer response.Body.Close()
if response.StatusCode != http.StatusOK && response.StatusCode != http.StatusCreated {
if response.StatusCode >= 400 && response.StatusCode < 500 && response.StatusCode != http.StatusRequestTimeout && response.StatusCode != http.StatusTooManyRequests {
return ports.ExecutionResult{Known: true}
}
return ports.ExecutionResult{}
}
receipt, err := decodeReceipt(response.Body)
if err != nil || !matchingReceipt(receipt, action) {
return ports.ExecutionResult{}
}
return ports.ExecutionResult{Known: true, Succeeded: true}
}
func (e *Executor) Reconcile(action commitcontract.AuthorizedAction) ports.ExecutionResult {
if _, err := e.requestFor(action); err != nil {
return ports.ExecutionResult{Known: true}
}
endpoint, err := e.endpoint(e.binding.ReceiptPath, string(action.Proposal.IntentID))
if err != nil {
return ports.ExecutionResult{Known: true}
}
request, err := http.NewRequest(http.MethodGet, endpoint, nil)
if err != nil {
return ports.ExecutionResult{Known: true}
}
request.Header.Set("Authorization", "Bearer "+e.secret)
request.Header.Set("Accept", "application/json")
response, err := e.client.Do(request)
if err != nil {
return ports.ExecutionResult{}
}
defer response.Body.Close()
if response.StatusCode != http.StatusOK {
return ports.ExecutionResult{}
}
receipt, err := decodeReceipt(response.Body)
if err != nil {
return ports.ExecutionResult{}
}
if !matchingReceipt(receipt, action) {
return ports.ExecutionResult{Known: true}
}
return ports.ExecutionResult{Known: true, Succeeded: true}
}
func (e *Executor) requestFor(action commitcontract.AuthorizedAction) (Payload, error) {
if action.Proposal.NotificationSubject.BindingID != e.binding.ID || action.Proposal.NotificationSubject.BindingRevision != e.binding.Revision ||
action.PrincipalID != e.binding.OwnerPrincipalID {
return Payload{}, ErrInvalidNotification
}
return PayloadFromAction(action)
}
func (e *Executor) endpoint(path, intent string) (string, error) {
base, err := url.Parse(e.binding.HTTPSOrigin)
if err != nil {
return "", err
}
base.Path = path + "/" + url.PathEscape(intent)
base.RawPath, base.RawQuery, base.Fragment = "", "", ""
return base.String(), nil
}
func decodeReceipt(reader io.Reader) (Receipt, error) {
var receipt Receipt
decoder := json.NewDecoder(io.LimitReader(reader, maxResponseBytes+1))
if err := decoder.Decode(&receipt); err != nil {
return Receipt{}, err
}
if receipt.ReceiptID == "" || receipt.ActionIntentID == "" || len(receipt.RequestDigest) != 64 || receipt.AcceptedAt <= 0 {
return Receipt{}, fmt.Errorf("invalid receipt")
}
return receipt, nil
}
func matchingReceipt(receipt Receipt, action commitcontract.AuthorizedAction) bool {
return receipt.ActionIntentID == action.Proposal.IntentID && receipt.RequestDigest == action.Proposal.NotificationSubject.RequestDigest
}
@@ -0,0 +1,109 @@
package notificationexecutor
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"git.cin.su/m/Assistant/contracts/capability"
commitcontract "git.cin.su/m/Assistant/contracts/commit"
"git.cin.su/m/Assistant/contracts/identity"
)
func testAction(t *testing.T) commitcontract.AuthorizedAction {
t.Helper()
payload := Payload{ActionIntentID: "intent-1", TemplateCode: commitcontract.ProspectiveDeliveredTemplateV1, ProspectiveItemID: "item-1", ExpectedRevision: 2, Title: "Домовой", Message: "Пора вернуться к задаче"}
digest, err := Digest(payload)
if err != nil {
t.Fatal(err)
}
return commitcontract.AuthorizedAction{PrincipalID: identity.PrincipalID("owner-1"), Proposal: commitcontract.ActionProposal{
IntentID: payload.ActionIntentID, CapabilityID: commitcontract.CapabilityProspectiveNotification, Operation: commitcontract.OperationDeliverProspectiveNotification,
NotificationSubject: commitcontract.ProspectiveNotificationSubject{ProspectiveItemID: "item-1", ExpectedRevision: 2, TemplateCode: payload.TemplateCode, BindingID: "binding-1", BindingRevision: 1, RenderedTitle: payload.Title, RenderedMessage: payload.Message, RequestDigest: digest},
}}
}
func testBinding(origin string) capability.Binding {
return capability.Binding{ID: "binding-1", CapabilityID: commitcontract.CapabilityProspectiveNotification, OwnerPrincipalID: "owner-1", Revision: 1, State: capability.BindingActive, HTTPSOrigin: origin, DeliveryPath: "/delivery/v1/deliveries", ReceiptPath: "/delivery/v1/receipts", Protocol: capability.ProtocolIdempotentReceiptV1, SecretRef: "write-v1", VerifiedAtUnixMillis: 1, Health: capability.HealthAvailable}
}
func TestExecutorPutAndReconcileMatchReceipt(t *testing.T) {
action := testAction(t)
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "Bearer secret" {
t.Error("missing credential")
}
if r.Method == http.MethodPut && r.URL.Path != "/delivery/v1/deliveries/intent-1" {
t.Errorf("bad put path %s", r.URL.Path)
}
if r.Method == http.MethodGet && r.URL.Path != "/delivery/v1/receipts/intent-1" {
t.Errorf("bad get path %s", r.URL.Path)
}
_ = json.NewEncoder(w).Encode(Receipt{ReceiptID: "receipt-intent-1", ActionIntentID: "intent-1", RequestDigest: action.Proposal.NotificationSubject.RequestDigest, AcceptedAt: 1})
}))
defer server.Close()
executor, err := New(testBinding(server.URL), "secret", server.Client())
if err != nil {
t.Fatal(err)
}
if result := executor.Execute(action); !result.Known || !result.Succeeded {
t.Fatalf("execute: %+v", result)
}
if result := executor.Reconcile(action); !result.Known || !result.Succeeded {
t.Fatalf("reconcile: %+v", result)
}
}
func TestExecutorAmbiguousAndIntegrityOutcomes(t *testing.T) {
action := testAction(t)
for name, handler := range map[string]http.HandlerFunc{
"server error": func(w http.ResponseWriter, _ *http.Request) { http.Error(w, "no", http.StatusBadGateway) },
"bad receipt": func(w http.ResponseWriter, _ *http.Request) {
_ = json.NewEncoder(w).Encode(Receipt{ReceiptID: "r", ActionIntentID: "other", RequestDigest: action.Proposal.NotificationSubject.RequestDigest, AcceptedAt: 1})
},
} {
t.Run(name, func(t *testing.T) {
server := httptest.NewTLSServer(handler)
defer server.Close()
executor, err := New(testBinding(server.URL), "secret", server.Client())
if err != nil {
t.Fatal(err)
}
if result := executor.Execute(action); result.Known {
t.Fatalf("must be unknown: %+v", result)
}
})
}
action.Proposal.NotificationSubject.RequestDigest = "bad"
server := httptest.NewTLSServer(http.NotFoundHandler())
defer server.Close()
executor, err := New(testBinding(server.URL), "secret", server.Client())
if err != nil {
t.Fatal(err)
}
if result := executor.Execute(action); !result.Known || result.Succeeded {
t.Fatalf("invalid evidence must fail closed: %+v", result)
}
}
func TestExecutorDoesNotFollowRedirect(t *testing.T) {
action := testAction(t)
targetCalls := 0
target := httptest.NewTLSServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { targetCalls++ }))
defer target.Close()
redirect := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, target.URL, http.StatusTemporaryRedirect)
}))
defer redirect.Close()
executor, err := New(testBinding(redirect.URL), "secret", redirect.Client())
if err != nil {
t.Fatal(err)
}
if result := executor.Execute(action); result.Known {
t.Fatalf("redirect must be ambiguous: %+v", result)
}
if targetCalls != 0 {
t.Fatal("redirect target was called")
}
}
@@ -6,10 +6,13 @@ import (
"fmt"
"git.cin.su/m/Assistant/contracts/capability"
"git.cin.su/m/Assistant/contracts/identity"
"github.com/jackc/pgx/v5/pgxpool"
)
var ErrCapabilityHealthNotFound = errors.New("capability health not found")
var ErrCapabilityBindingNotFound = errors.New("capability binding not found")
var ErrCapabilityBindingConflict = errors.New("capability binding conflict")
type CapabilityRepository struct{ pool *pgxpool.Pool }
@@ -56,3 +59,72 @@ func (r *CapabilityRepository) RebuildHealthProjection(ctx context.Context) (int
}
return result.RowsAffected(), nil
}
func (r *CapabilityRepository) SaveBinding(ctx context.Context, binding capability.Binding) error {
tx, err := r.pool.Begin(ctx)
if err != nil {
return err
}
defer func() { _ = tx.Rollback(ctx) }()
result, err := tx.Exec(ctx, `INSERT INTO capability.notification_bindings (
binding_id,capability_id,owner_principal_id,revision,state,https_origin,delivery_path,receipt_path,
protocol,secret_ref,verified_at_unix_millis,health)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) ON CONFLICT DO NOTHING`,
binding.ID, binding.CapabilityID, binding.OwnerPrincipalID, binding.Revision, binding.State,
binding.HTTPSOrigin, binding.DeliveryPath, binding.ReceiptPath, binding.Protocol, binding.SecretRef,
binding.VerifiedAtUnixMillis, binding.Health)
if err != nil {
return err
}
if result.RowsAffected() == 0 {
existing, loadErr := r.LoadBinding(ctx, binding.ID, binding.Revision)
if loadErr != nil || existing != binding {
return ErrCapabilityBindingConflict
}
}
if binding.State == capability.BindingActive {
result, err = tx.Exec(ctx, `INSERT INTO capability.notification_binding_heads (capability_id,owner_principal_id,binding_id,revision)
VALUES ($1,$2,$3,$4) ON CONFLICT (capability_id,owner_principal_id) DO UPDATE SET binding_id=EXCLUDED.binding_id,revision=EXCLUDED.revision
WHERE capability.notification_binding_heads.revision < EXCLUDED.revision`, binding.CapabilityID, binding.OwnerPrincipalID, binding.ID, binding.Revision)
if err != nil || result.RowsAffected() != 1 {
if err != nil {
return err
}
return ErrCapabilityBindingConflict
}
} else {
_, err = tx.Exec(ctx, `DELETE FROM capability.notification_binding_heads WHERE capability_id=$1 AND owner_principal_id=$2 AND binding_id=$3 AND revision < $4`, binding.CapabilityID, binding.OwnerPrincipalID, binding.ID, binding.Revision)
if err != nil {
return err
}
}
return tx.Commit(ctx)
}
func (r *CapabilityRepository) LoadBinding(ctx context.Context, id capability.CapabilityBindingID, revision int64) (capability.Binding, error) {
var binding capability.Binding
err := r.pool.QueryRow(ctx, `SELECT binding_id,capability_id,owner_principal_id,revision,state,https_origin,delivery_path,receipt_path,
protocol,secret_ref,verified_at_unix_millis,health FROM capability.notification_bindings WHERE binding_id=$1 AND revision=$2`, id, revision).
Scan(&binding.ID, &binding.CapabilityID, &binding.OwnerPrincipalID, &binding.Revision, &binding.State,
&binding.HTTPSOrigin, &binding.DeliveryPath, &binding.ReceiptPath, &binding.Protocol, &binding.SecretRef,
&binding.VerifiedAtUnixMillis, &binding.Health)
if err != nil {
return capability.Binding{}, fmt.Errorf("load capability binding: %w", err)
}
return binding, nil
}
func (r *CapabilityRepository) LoadActiveBinding(ctx context.Context, capabilityID capability.CapabilityID, owner identity.PrincipalID) (capability.Binding, error) {
var binding capability.Binding
err := r.pool.QueryRow(ctx, `SELECT b.binding_id,b.capability_id,b.owner_principal_id,b.revision,b.state,b.https_origin,b.delivery_path,b.receipt_path,
b.protocol,b.secret_ref,b.verified_at_unix_millis,b.health FROM capability.notification_binding_heads h
JOIN capability.notification_bindings b ON b.binding_id=h.binding_id AND b.revision=h.revision
WHERE h.capability_id=$1 AND h.owner_principal_id=$2 AND b.state=$3`, capabilityID, owner, capability.BindingActive).
Scan(&binding.ID, &binding.CapabilityID, &binding.OwnerPrincipalID, &binding.Revision, &binding.State,
&binding.HTTPSOrigin, &binding.DeliveryPath, &binding.ReceiptPath, &binding.Protocol, &binding.SecretRef,
&binding.VerifiedAtUnixMillis, &binding.Health)
if err != nil {
return capability.Binding{}, fmt.Errorf("load active capability binding: %w", err)
}
return binding, nil
}
+31 -9
View File
@@ -7,6 +7,7 @@ import (
"reflect"
commitcontract "git.cin.su/m/Assistant/contracts/commit"
"git.cin.su/m/Assistant/contracts/continuity"
"git.cin.su/m/Assistant/contracts/identity"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
@@ -30,6 +31,12 @@ func OpenCommit(ctx context.Context, c ContinuityConfig) (*CommitRepository, err
}
func (r *CommitRepository) Close() { r.pool.Close() }
func (r *CommitRepository) SaveAuthorized(ctx context.Context, a commitcontract.AuthorizedAction) error {
prospectiveItemID := a.Proposal.ContinuitySubject.ProspectiveItemID
expectedRevision := a.Proposal.ContinuitySubject.ExpectedRevision
if a.Proposal.Operation == commitcontract.OperationDeliverProspectiveNotification {
prospectiveItemID = a.Proposal.NotificationSubject.ProspectiveItemID
expectedRevision = a.Proposal.NotificationSubject.ExpectedRevision
}
tx, err := r.pool.Begin(ctx)
if err != nil {
return err
@@ -42,15 +49,19 @@ func (r *CommitRepository) SaveAuthorized(ctx context.Context, a commitcontract.
authorized_action_id,action_intent_id,actor_id,principal_id,grant_id,grant_revision,scope_id,
task_id,goal_id,task_scope_id,task_revision,capability_id,operation,prospective_item_id,
expected_prospective_revision,prospective_from_state,prospective_to_state,evidence_ref,
proposal_state,action_state,authorized_at_unix_millis,envelope_revision)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,1)
proposal_state,action_state,authorized_at_unix_millis,envelope_revision,notification_template_code,
capability_binding_id,capability_binding_revision,notification_title,notification_message,request_digest)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,1,$22,$23,$24,$25,$26,$27)
ON CONFLICT DO NOTHING`,
a.ID, a.Proposal.IntentID, a.Proposal.ActorID, a.PrincipalID, a.GrantID, a.GrantRevision, a.ScopeID,
a.Proposal.Target.TaskID, a.Proposal.Target.GoalID, a.Proposal.Target.ScopeID, a.Proposal.Target.Revision,
a.Proposal.CapabilityID, a.Proposal.Operation, a.Proposal.ContinuitySubject.ProspectiveItemID,
a.Proposal.ContinuitySubject.ExpectedRevision, a.Proposal.ContinuitySubject.FromState,
a.Proposal.CapabilityID, a.Proposal.Operation, prospectiveItemID,
expectedRevision, a.Proposal.ContinuitySubject.FromState,
a.Proposal.ContinuitySubject.ToState, a.Proposal.ContinuitySubject.EvidenceRef,
a.Proposal.State, a.State, a.AuthorizedAtUnixMillis)
a.Proposal.State, a.State, a.AuthorizedAtUnixMillis, a.Proposal.NotificationSubject.TemplateCode,
a.Proposal.NotificationSubject.BindingID, a.Proposal.NotificationSubject.BindingRevision,
a.Proposal.NotificationSubject.RenderedTitle, a.Proposal.NotificationSubject.RenderedMessage,
a.Proposal.NotificationSubject.RequestDigest)
if err != nil {
return err
}
@@ -128,7 +139,8 @@ func (r *CommitRepository) ListAuthorizedByOwner(ctx context.Context, actorID id
rows, err := r.pool.Query(ctx, `SELECT authorized_action_id,action_intent_id,actor_id,principal_id,grant_id,grant_revision,scope_id,
task_id,goal_id,task_scope_id,task_revision,capability_id,operation,prospective_item_id,
expected_prospective_revision,prospective_from_state,prospective_to_state,evidence_ref,
proposal_state,action_state,authorized_at_unix_millis
proposal_state,action_state,authorized_at_unix_millis,notification_template_code,capability_binding_id,
capability_binding_revision,notification_title,notification_message,request_digest
FROM commit.authorized_action_envelopes
WHERE actor_id=$1 AND principal_id=$2
ORDER BY authorized_at_unix_millis DESC, authorized_action_id DESC LIMIT $3`, actorID, principalID, limit)
@@ -163,7 +175,8 @@ func loadAuthorized(ctx context.Context, querier commitRowQuerier, id commitcont
err := scanAuthorized(querier.QueryRow(ctx, `SELECT authorized_action_id,action_intent_id,actor_id,principal_id,grant_id,grant_revision,scope_id,
task_id,goal_id,task_scope_id,task_revision,capability_id,operation,prospective_item_id,
expected_prospective_revision,prospective_from_state,prospective_to_state,evidence_ref,
proposal_state,action_state,authorized_at_unix_millis
proposal_state,action_state,authorized_at_unix_millis,notification_template_code,capability_binding_id,
capability_binding_revision,notification_title,notification_message,request_digest
FROM commit.authorized_action_envelopes WHERE authorized_action_id=$1`, id), &action)
if errors.Is(err, pgx.ErrNoRows) {
return commitcontract.AuthorizedAction{}, ErrAuthorizedActionNotFound
@@ -175,12 +188,21 @@ func loadAuthorized(ctx context.Context, querier commitRowQuerier, id commitcont
}
func scanAuthorized(row commitRowScanner, action *commitcontract.AuthorizedAction) error {
return row.Scan(
err := row.Scan(
&action.ID, &action.Proposal.IntentID, &action.Proposal.ActorID, &action.PrincipalID,
&action.GrantID, &action.GrantRevision, &action.ScopeID, &action.Proposal.Target.TaskID,
&action.Proposal.Target.GoalID, &action.Proposal.Target.ScopeID, &action.Proposal.Target.Revision,
&action.Proposal.CapabilityID, &action.Proposal.Operation, &action.Proposal.ContinuitySubject.ProspectiveItemID,
&action.Proposal.ContinuitySubject.ExpectedRevision, &action.Proposal.ContinuitySubject.FromState,
&action.Proposal.ContinuitySubject.ToState, &action.Proposal.ContinuitySubject.EvidenceRef,
&action.Proposal.State, &action.State, &action.AuthorizedAtUnixMillis)
&action.Proposal.State, &action.State, &action.AuthorizedAtUnixMillis,
&action.Proposal.NotificationSubject.TemplateCode, &action.Proposal.NotificationSubject.BindingID,
&action.Proposal.NotificationSubject.BindingRevision, &action.Proposal.NotificationSubject.RenderedTitle,
&action.Proposal.NotificationSubject.RenderedMessage, &action.Proposal.NotificationSubject.RequestDigest)
if err == nil && action.Proposal.Operation == commitcontract.OperationDeliverProspectiveNotification {
action.Proposal.NotificationSubject.ProspectiveItemID = action.Proposal.ContinuitySubject.ProspectiveItemID
action.Proposal.NotificationSubject.ExpectedRevision = action.Proposal.ContinuitySubject.ExpectedRevision
action.Proposal.ContinuitySubject = continuity.SatisfyDeliveredSubject{}
}
return err
}
+38
View File
@@ -17,6 +17,7 @@ import (
"log"
"net/http"
"os"
"strconv"
"strings"
"time"
)
@@ -183,6 +184,43 @@ func main() {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(status)
})
mux.HandleFunc("GET /v1/capability-bindings/{id}/revisions/{revision}", func(w http.ResponseWriter, r *http.Request) {
revision, err := strconv.ParseInt(r.PathValue("revision"), 10, 64)
if err != nil || revision < 1 {
http.Error(w, "invalid binding revision", http.StatusBadRequest)
return
}
binding, err := capabilityRepository.LoadBinding(r.Context(), capability.CapabilityBindingID(r.PathValue("id")), revision)
if err != nil {
http.Error(w, "capability binding not found", http.StatusNotFound)
return
}
if err = capabilitydomain.ValidateBinding(binding); err != nil {
http.Error(w, "capability binding invalid", http.StatusConflict)
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(binding)
})
mux.HandleFunc("GET /v1/capability-bindings/active", func(w http.ResponseWriter, r *http.Request) {
capabilityID := capability.CapabilityID(r.URL.Query().Get("capability_id"))
ownerID := identity.PrincipalID(r.URL.Query().Get("owner_principal_id"))
if capabilityID == "" || ownerID == "" {
http.Error(w, "invalid binding query", http.StatusBadRequest)
return
}
binding, err := capabilityRepository.LoadActiveBinding(r.Context(), capabilityID, ownerID)
if err != nil {
http.Error(w, "active capability binding not found", http.StatusNotFound)
return
}
if err = capabilitydomain.UsableBinding(binding); err != nil {
http.Error(w, "active capability binding unavailable", http.StatusConflict)
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(binding)
})
mux.HandleFunc("POST /v1/capabilities/health-projection/rebuild", func(w http.ResponseWriter, r *http.Request) {
rebuilt, err := capabilityRepository.RebuildHealthProjection(r.Context())
if err != nil {
+52
View File
@@ -0,0 +1,52 @@
package main
import (
"fmt"
"log"
"net/http"
"os"
"strings"
"time"
)
func main() {
addr := envOr("LISTEN_ADDR", ":8085")
writeToken, err := readSecretFile("RECEIVER_WRITE_TOKEN_FILE")
if err != nil {
log.Fatal(err)
}
readToken, err := readSecretFile("RECEIVER_READ_TOKEN_FILE")
if err != nil {
log.Fatal(err)
}
server, err := newReceiverServer(envOr("RECEIPT_DIRECTORY", "/var/lib/notification-receiver"), writeToken, readToken, time.Now)
if err != nil {
log.Fatal(err)
}
log.Printf("notification-receiver listening on %s", addr)
if err = http.ListenAndServe(addr, server.routes()); err != nil {
log.Fatal(err)
}
}
func readSecretFile(name string) (string, error) {
path := os.Getenv(name)
if path == "" {
return "", fmt.Errorf("%s is required", name)
}
value, err := os.ReadFile(path)
if err != nil {
return "", fmt.Errorf("read %s: %w", name, err)
}
secret := strings.TrimSpace(string(value))
if secret == "" {
return "", fmt.Errorf("%s is empty", name)
}
return secret, nil
}
func envOr(name, fallback string) string {
if value := os.Getenv(name); value != "" {
return value
}
return fallback
}
+246
View File
@@ -0,0 +1,246 @@
package main
import (
"crypto/subtle"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"os"
"path/filepath"
"regexp"
"sort"
"strconv"
"strings"
"time"
"git.cin.su/m/Assistant/adapters/notificationexecutor"
)
const maxDeliveryBytes = 8 << 10
var safeIntent = regexp.MustCompile(`^[A-Za-z0-9_-]{1,128}$`)
type storedReceipt struct {
notificationexecutor.Receipt
Payload notificationexecutor.Payload `json:"payload"`
}
type receiverServer struct {
directory string
writeToken string
readToken string
now func() time.Time
}
func newReceiverServer(directory, writeToken, readToken string, now func() time.Time) (*receiverServer, error) {
if directory == "" || strings.TrimSpace(writeToken) == "" || strings.TrimSpace(readToken) == "" ||
writeToken == readToken || now == nil {
return nil, errors.New("invalid receiver configuration")
}
if err := os.MkdirAll(directory, 0700); err != nil {
return nil, fmt.Errorf("create receipt directory: %w", err)
}
return &receiverServer{directory: directory, writeToken: writeToken, readToken: readToken, now: now}, nil
}
func (s *receiverServer) routes() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = io.WriteString(w, `{"status":"ok"}`)
})
mux.HandleFunc("PUT /delivery/v1/deliveries/{intent}", s.put)
mux.HandleFunc("GET /delivery/v1/receipts/{intent}", s.get)
mux.HandleFunc("GET /delivery/v1/receipts", s.list)
return mux
}
func (s *receiverServer) put(w http.ResponseWriter, r *http.Request) {
if !tokenMatches(s.writeToken, r) {
http.Error(w, "receiver writer denied", http.StatusUnauthorized)
return
}
intent := r.PathValue("intent")
if !safeIntent.MatchString(intent) {
http.Error(w, "invalid action intent", http.StatusBadRequest)
return
}
var request notificationexecutor.DeliveryRequest
decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxDeliveryBytes))
decoder.DisallowUnknownFields()
if err := decoder.Decode(&request); err != nil || string(request.ActionIntentID) != intent {
http.Error(w, "invalid delivery", http.StatusBadRequest)
return
}
if err := ensureEOF(decoder); err != nil {
http.Error(w, "invalid delivery", http.StatusBadRequest)
return
}
digest, err := notificationexecutor.Digest(request.Payload)
if err != nil || digest != request.RequestDigest {
http.Error(w, "delivery digest mismatch", http.StatusConflict)
return
}
receipt := storedReceipt{Receipt: notificationexecutor.Receipt{ReceiptID: "receipt-" + intent, ActionIntentID: request.ActionIntentID, RequestDigest: digest, AcceptedAt: s.now().UnixMilli()}, Payload: request.Payload}
created, existing, err := s.create(receipt)
if err != nil {
http.Error(w, "receipt unavailable", http.StatusInternalServerError)
return
}
if !created {
if existing.RequestDigest != digest {
http.Error(w, "action intent digest conflict", http.StatusConflict)
return
}
receipt = existing
}
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Cache-Control", "no-store")
if created {
w.WriteHeader(http.StatusCreated)
}
_ = json.NewEncoder(w).Encode(receipt)
}
func (s *receiverServer) get(w http.ResponseWriter, r *http.Request) {
if !tokenMatches(s.writeToken, r) && !tokenMatches(s.readToken, r) {
http.Error(w, "receiver reader denied", http.StatusUnauthorized)
return
}
intent := r.PathValue("intent")
if !safeIntent.MatchString(intent) {
http.Error(w, "invalid action intent", http.StatusBadRequest)
return
}
receipt, err := s.load(intent)
if errors.Is(err, os.ErrNotExist) {
http.Error(w, "receipt not found", http.StatusNotFound)
return
}
if err != nil {
http.Error(w, "receipt unavailable", http.StatusInternalServerError)
return
}
writeStoredReceipt(w, receipt)
}
func (s *receiverServer) list(w http.ResponseWriter, r *http.Request) {
if !tokenMatches(s.readToken, r) {
http.Error(w, "receiver reader denied", http.StatusUnauthorized)
return
}
limit := 50
if raw := r.URL.Query().Get("limit"); raw != "" {
parsed, err := strconv.Atoi(raw)
if err != nil || parsed < 1 || parsed > 100 {
http.Error(w, "invalid limit", http.StatusBadRequest)
return
}
limit = parsed
}
entries, err := os.ReadDir(s.directory)
if err != nil {
http.Error(w, "receipts unavailable", http.StatusInternalServerError)
return
}
receipts := make([]storedReceipt, 0, len(entries))
for _, entry := range entries {
if entry.IsDir() || filepath.Ext(entry.Name()) != ".json" {
continue
}
receipt, loadErr := s.load(strings.TrimSuffix(entry.Name(), ".json"))
if loadErr != nil {
continue
}
receipts = append(receipts, receipt)
}
sort.Slice(receipts, func(i, j int) bool {
if receipts[i].AcceptedAt == receipts[j].AcceptedAt {
return receipts[i].ReceiptID > receipts[j].ReceiptID
}
return receipts[i].AcceptedAt > receipts[j].AcceptedAt
})
if len(receipts) > limit {
receipts = receipts[:limit]
}
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Cache-Control", "no-store")
_ = json.NewEncoder(w).Encode(struct {
Items []storedReceipt `json:"items"`
}{Items: receipts})
}
func (s *receiverServer) create(receipt storedReceipt) (bool, storedReceipt, error) {
encoded, err := json.Marshal(receipt)
if err != nil {
return false, storedReceipt{}, err
}
temporary, err := os.CreateTemp(s.directory, ".receipt-*.tmp")
if err != nil {
return false, storedReceipt{}, err
}
temporaryName := temporary.Name()
defer os.Remove(temporaryName)
if err = temporary.Chmod(0600); err == nil {
_, err = temporary.Write(encoded)
}
if err == nil {
err = temporary.Sync()
}
closeErr := temporary.Close()
if err == nil {
err = closeErr
}
if err != nil {
return false, storedReceipt{}, err
}
target := s.path(string(receipt.ActionIntentID))
if err = os.Link(temporaryName, target); err == nil {
return true, storedReceipt{}, nil
}
if !errors.Is(err, os.ErrExist) {
return false, storedReceipt{}, err
}
existing, err := s.load(string(receipt.ActionIntentID))
return false, existing, err
}
func (s *receiverServer) load(intent string) (storedReceipt, error) {
var receipt storedReceipt
file, err := os.Open(s.path(intent))
if err != nil {
return receipt, err
}
defer file.Close()
decoder := json.NewDecoder(io.LimitReader(file, maxDeliveryBytes+1))
if err = decoder.Decode(&receipt); err != nil {
return storedReceipt{}, err
}
if string(receipt.ActionIntentID) != intent {
return storedReceipt{}, errors.New("receipt identity mismatch")
}
return receipt, nil
}
func (s *receiverServer) path(intent string) string {
return filepath.Join(s.directory, intent+".json")
}
func tokenMatches(expected string, r *http.Request) bool {
provided := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
return len(provided) == len(expected) && subtle.ConstantTimeCompare([]byte(provided), []byte(expected)) == 1
}
func ensureEOF(decoder *json.Decoder) error {
var extra any
err := decoder.Decode(&extra)
if errors.Is(err, io.EOF) {
return nil
}
return errors.New("trailing JSON")
}
func writeStoredReceipt(w http.ResponseWriter, receipt storedReceipt) {
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Cache-Control", "no-store")
_ = json.NewEncoder(w).Encode(receipt)
}
+122
View File
@@ -0,0 +1,122 @@
package main
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"git.cin.su/m/Assistant/adapters/notificationexecutor"
commitcontract "git.cin.su/m/Assistant/contracts/commit"
)
func delivery(t *testing.T, intent, message string) notificationexecutor.DeliveryRequest {
t.Helper()
payload := notificationexecutor.Payload{ActionIntentID: commitcontract.ActionIntentID(intent), TemplateCode: commitcontract.ProspectiveDeliveredTemplateV1, ProspectiveItemID: "item-1", ExpectedRevision: 1, Title: "Домовой", Message: message}
digest, err := notificationexecutor.Digest(payload)
if err != nil {
t.Fatal(err)
}
return notificationexecutor.DeliveryRequest{Payload: payload, RequestDigest: digest}
}
func request(t *testing.T, handler http.Handler, method, path, token string, value any) *httptest.ResponseRecorder {
t.Helper()
var body bytes.Buffer
if value != nil {
if err := json.NewEncoder(&body).Encode(value); err != nil {
t.Fatal(err)
}
}
r := httptest.NewRequest(method, path, &body)
if token != "" {
r.Header.Set("Authorization", "Bearer "+token)
}
w := httptest.NewRecorder()
handler.ServeHTTP(w, r)
return w
}
func TestReceiverIdempotencyConflictAndRestart(t *testing.T) {
directory := t.TempDir()
clock := func() time.Time { return time.UnixMilli(100) }
server, err := newReceiverServer(directory, "write-secret", "read-secret", clock)
if err != nil {
t.Fatal(err)
}
first := delivery(t, "intent-1", "message one")
w := request(t, server.routes(), http.MethodPut, "/delivery/v1/deliveries/intent-1", "write-secret", first)
if w.Code != http.StatusCreated {
t.Fatalf("create status %d: %s", w.Code, w.Body.String())
}
w = request(t, server.routes(), http.MethodPut, "/delivery/v1/deliveries/intent-1", "write-secret", first)
if w.Code != http.StatusOK {
t.Fatalf("repeat status %d", w.Code)
}
conflict := delivery(t, "intent-1", "different")
w = request(t, server.routes(), http.MethodPut, "/delivery/v1/deliveries/intent-1", "write-secret", conflict)
if w.Code != http.StatusConflict {
t.Fatalf("conflict status %d", w.Code)
}
restarted, err := newReceiverServer(directory, "write-secret", "read-secret", clock)
if err != nil {
t.Fatal(err)
}
w = request(t, restarted.routes(), http.MethodGet, "/delivery/v1/receipts/intent-1", "write-secret", nil)
if w.Code != http.StatusOK {
t.Fatalf("restart get status %d", w.Code)
}
w = request(t, restarted.routes(), http.MethodGet, "/delivery/v1/receipts?limit=10", "read-secret", nil)
if w.Code != http.StatusOK {
t.Fatalf("list status %d", w.Code)
}
}
func TestReceiverSeparatesReadAndWriteAuthority(t *testing.T) {
server, err := newReceiverServer(t.TempDir(), "write-secret", "read-secret", time.Now)
if err != nil {
t.Fatal(err)
}
handler := server.routes()
value := delivery(t, "intent-1", "message")
if w := request(t, handler, http.MethodPut, "/delivery/v1/deliveries/intent-1", "read-secret", value); w.Code != http.StatusUnauthorized {
t.Fatalf("reader wrote: %d", w.Code)
}
if w := request(t, handler, http.MethodGet, "/delivery/v1/receipts", "write-secret", nil); w.Code != http.StatusUnauthorized {
t.Fatalf("writer listed: %d", w.Code)
}
}
func TestConcurrentDuplicateCreatesOneReceipt(t *testing.T) {
server, err := newReceiverServer(t.TempDir(), "write-secret", "read-secret", time.Now)
if err != nil {
t.Fatal(err)
}
value := delivery(t, "intent-concurrent", "message")
handler := server.routes()
codes := make(chan int, 8)
var group sync.WaitGroup
for range 8 {
group.Add(1)
go func() {
defer group.Done()
codes <- request(t, handler, http.MethodPut, "/delivery/v1/deliveries/intent-concurrent", "write-secret", value).Code
}()
}
group.Wait()
close(codes)
created := 0
for code := range codes {
if code == http.StatusCreated {
created++
} else if code != http.StatusOK {
t.Fatalf("unexpected status %d", code)
}
}
if created != 1 {
t.Fatalf("created=%d", created)
}
}
+31 -1
View File
@@ -1,10 +1,40 @@
package capability
import "git.cin.su/m/Assistant/platform/ids"
import (
"git.cin.su/m/Assistant/contracts/identity"
"git.cin.su/m/Assistant/platform/ids"
)
type CapabilityID ids.ID
type CapabilityBindingID ids.ID
type BindingState string
type BindingProtocol string
const (
BindingActive BindingState = "ACTIVE"
BindingRetired BindingState = "RETIRED"
ProtocolIdempotentReceiptV1 BindingProtocol = "IDEMPOTENT_RECEIPT_V1"
)
// Binding is Capability-owned correctness state. SecretRef is an opaque
// lookup name; it must never contain credential bytes.
type Binding struct {
ID CapabilityBindingID
CapabilityID CapabilityID
OwnerPrincipalID identity.PrincipalID
Revision int64
State BindingState
HTTPSOrigin string
DeliveryPath string
ReceiptPath string
Protocol BindingProtocol
SecretRef string
VerifiedAtUnixMillis int64
Health HealthState
}
type HealthState string
const (
+29 -11
View File
@@ -28,20 +28,38 @@ const (
type ExternalOperation string
const (
CapabilityTestExecutor capability.CapabilityID = "test-executor"
CapabilityContinuityTransition capability.CapabilityID = "continuity-transition"
OperationTestIdempotentEffect ExternalOperation = "test-idempotent-effect"
OperationSatisfyDelivered ExternalOperation = "satisfy-delivered"
CapabilityTestExecutor capability.CapabilityID = "test-executor"
CapabilityContinuityTransition capability.CapabilityID = "continuity-transition"
CapabilityProspectiveNotification capability.CapabilityID = "prospective-notification-delivery"
OperationTestIdempotentEffect ExternalOperation = "test-idempotent-effect"
OperationSatisfyDelivered ExternalOperation = "satisfy-delivered"
OperationDeliverProspectiveNotification ExternalOperation = "deliver-prospective-notification-v1"
)
const ProspectiveDeliveredTemplateV1 = "prospective-delivered-v1"
// ProspectiveNotificationSubject is immutable Commit evidence. Destination
// data is resolved from the exact Capability binding revision, never from UI.
type ProspectiveNotificationSubject struct {
ProspectiveItemID continuity.ProspectiveItemID
ExpectedRevision continuity.ProspectiveRevision
TemplateCode string
BindingID capability.CapabilityBindingID
BindingRevision int64
RenderedTitle string
RenderedMessage string
RequestDigest string
}
type ActionProposal struct {
IntentID ActionIntentID
ActorID identity.ActorID
CapabilityID capability.CapabilityID
Target work.TaskTarget
Operation ExternalOperation
ContinuitySubject continuity.SatisfyDeliveredSubject
State ActionState
IntentID ActionIntentID
ActorID identity.ActorID
CapabilityID capability.CapabilityID
Target work.TaskTarget
Operation ExternalOperation
ContinuitySubject continuity.SatisfyDeliveredSubject
NotificationSubject ProspectiveNotificationSubject
State ActionState
}
type AuthorizedAction struct {
+29 -1
View File
@@ -110,9 +110,20 @@ CREATE TABLE IF NOT EXISTS commit.authorized_action_envelopes (
action_state TEXT NOT NULL,
authorized_at_unix_millis BIGINT NOT NULL,
envelope_revision BIGINT NOT NULL,
notification_template_code TEXT NOT NULL DEFAULT '',
capability_binding_id TEXT NOT NULL DEFAULT '',
capability_binding_revision BIGINT NOT NULL DEFAULT 0,
notification_title TEXT NOT NULL DEFAULT '',
notification_message TEXT NOT NULL DEFAULT '',
request_digest TEXT NOT NULL DEFAULT '',
recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
CHECK (grant_revision > 0), CHECK (task_revision > 0),
CHECK (authorized_at_unix_millis > 0), CHECK (envelope_revision = 1)
CHECK (authorized_at_unix_millis > 0), CHECK (envelope_revision = 1),
CHECK (operation <> 'deliver-prospective-notification-v1' OR (
notification_template_code = 'prospective-delivered-v1' AND notification_title <> '' AND notification_message <> '' AND capability_binding_id <> '' AND
capability_binding_revision > 0 AND request_digest ~ '^[0-9a-f]{64}$' AND
expected_prospective_revision > 0 AND prospective_item_id <> ''
))
);
RESET ROLE;
@@ -157,6 +168,23 @@ CREATE TABLE IF NOT EXISTS capability.health_projection (
semantic_version INTEGER NOT NULL,
observed_at_unix_millis BIGINT
);
CREATE TABLE IF NOT EXISTS capability.notification_bindings (
binding_id TEXT NOT NULL, capability_id TEXT NOT NULL, owner_principal_id TEXT NOT NULL,
revision BIGINT NOT NULL, state TEXT NOT NULL, https_origin TEXT NOT NULL,
delivery_path TEXT NOT NULL, receipt_path TEXT NOT NULL, protocol TEXT NOT NULL,
secret_ref TEXT NOT NULL, verified_at_unix_millis BIGINT NOT NULL, health TEXT NOT NULL,
recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (binding_id, revision),
CHECK (revision > 0), CHECK (verified_at_unix_millis > 0),
CHECK (state IN ('ACTIVE','RETIRED')), CHECK (health IN ('AVAILABLE','DEGRADED','UNAVAILABLE')),
CHECK (protocol = 'IDEMPOTENT_RECEIPT_V1'), CHECK (https_origin LIKE 'https://%'),
CHECK (delivery_path LIKE '/%'), CHECK (receipt_path LIKE '/%')
);
CREATE TABLE IF NOT EXISTS capability.notification_binding_heads (
capability_id TEXT NOT NULL, owner_principal_id TEXT NOT NULL,
binding_id TEXT NOT NULL, revision BIGINT NOT NULL,
PRIMARY KEY (capability_id, owner_principal_id), CHECK (revision > 0)
);
RESET ROLE;
SET ROLE continuity_role;
@@ -0,0 +1,36 @@
BEGIN;
SET ROLE capability_role;
CREATE TABLE IF NOT EXISTS capability.notification_bindings (
binding_id TEXT NOT NULL,
capability_id TEXT NOT NULL,
owner_principal_id TEXT NOT NULL,
revision BIGINT NOT NULL,
state TEXT NOT NULL,
https_origin TEXT NOT NULL,
delivery_path TEXT NOT NULL,
receipt_path TEXT NOT NULL,
protocol TEXT NOT NULL,
secret_ref TEXT NOT NULL,
verified_at_unix_millis BIGINT NOT NULL,
health TEXT NOT NULL,
recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (binding_id, revision),
CHECK (revision > 0),
CHECK (verified_at_unix_millis > 0),
CHECK (state IN ('ACTIVE','RETIRED')),
CHECK (health IN ('AVAILABLE','DEGRADED','UNAVAILABLE')),
CHECK (protocol = 'IDEMPOTENT_RECEIPT_V1'),
CHECK (https_origin LIKE 'https://%'),
CHECK (delivery_path LIKE '/%'),
CHECK (receipt_path LIKE '/%')
);
CREATE TABLE IF NOT EXISTS capability.notification_binding_heads (
capability_id TEXT NOT NULL,
owner_principal_id TEXT NOT NULL,
binding_id TEXT NOT NULL,
revision BIGINT NOT NULL,
PRIMARY KEY (capability_id, owner_principal_id),
CHECK (revision > 0)
);
RESET ROLE;
COMMIT;
@@ -0,0 +1,22 @@
BEGIN;
SET ROLE commit_role;
ALTER TABLE commit.authorized_action_envelopes ADD COLUMN IF NOT EXISTS notification_template_code TEXT NOT NULL DEFAULT '';
ALTER TABLE commit.authorized_action_envelopes ADD COLUMN IF NOT EXISTS capability_binding_id TEXT NOT NULL DEFAULT '';
ALTER TABLE commit.authorized_action_envelopes ADD COLUMN IF NOT EXISTS capability_binding_revision BIGINT NOT NULL DEFAULT 0;
ALTER TABLE commit.authorized_action_envelopes ADD COLUMN IF NOT EXISTS notification_title TEXT NOT NULL DEFAULT '';
ALTER TABLE commit.authorized_action_envelopes ADD COLUMN IF NOT EXISTS notification_message TEXT NOT NULL DEFAULT '';
ALTER TABLE commit.authorized_action_envelopes ADD COLUMN IF NOT EXISTS request_digest TEXT NOT NULL DEFAULT '';
DO $$ BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname='authorized_action_notification_evidence_v1' AND conrelid='commit.authorized_action_envelopes'::regclass) THEN
ALTER TABLE commit.authorized_action_envelopes ADD CONSTRAINT authorized_action_notification_evidence_v1 CHECK (
operation <> 'deliver-prospective-notification-v1' OR (
notification_template_code = 'prospective-delivered-v1' AND notification_title <> '' AND notification_message <> '' AND
capability_binding_id <> '' AND capability_binding_revision > 0 AND
request_digest ~ '^[0-9a-f]{64}$' AND expected_prospective_revision > 0 AND prospective_item_id <> ''
)
) NOT VALID;
END IF;
END $$;
ALTER TABLE commit.authorized_action_envelopes VALIDATE CONSTRAINT authorized_action_notification_evidence_v1;
RESET ROLE;
COMMIT;
+2 -2
View File
@@ -1,5 +1,5 @@
{
"task_file": "dev/tasks/ARCH-000006-accept-prospective-notification-delivery.json",
"task_id": "ARCH-000006",
"task_file": "dev/tasks/DEV-000041-notification-delivery-foundation.json",
"task_id": "DEV-000041",
"task_revision": 1
}
@@ -0,0 +1,68 @@
{
"task_id": "DEV-000041",
"task_revision": 1,
"architecture_revision": "ADR-0005",
"base_repo_revision": "45fb5f8709ef9f327c5e41e7f9adb0d7a7d06866",
"risk_tier": "E",
"intent": "Implement the non-user-facing foundation for ADR-0005: Capability-owned versioned receiver binding, typed Commit notification evidence, a durable idempotent receipt receiver, and a strict HTTPS executor/reconciler adapter without enabling a Creator mutation route or installing credentials.",
"requirement_ids": ["INV-AUTH-001", "INV-OWN-001", "INV-TOPO-001", "ADR-0005"],
"allowed_change_scope": {
"modules": ["capability", "commit"],
"paths": [
"contracts/capability/types.go",
"contracts/commit/types.go",
"modules/capability/domain/binding.go",
"modules/capability/domain/binding_test.go",
"modules/capability/ports/repository.go",
"adapters/postgres/capability_repository.go",
"adapters/postgres/capability_repository_test.go",
"adapters/postgres/commit_repository.go",
"adapters/notificationexecutor/executor.go",
"adapters/notificationexecutor/executor_test.go",
"cmd/core-runtime/main.go",
"cmd/core-runtime/main_test.go",
"cmd/notification-receiver/main.go",
"cmd/notification-receiver/server.go",
"cmd/notification-receiver/server_test.go",
"database/migrations/010_capability_notification_binding.sql",
"database/migrations/011_commit_notification_envelope.sql",
"database/bootstrap/001_roles_and_schemas.sql",
"dev/ACTIVE_TASK.json",
"dev/tasks/DEV-000041-notification-delivery-foundation.json"
]
},
"forbidden_change_scope": {
"paths": [
"architecture/**",
"ci/**",
".gitea/**",
"modules/identity/**",
"modules/authority/**",
"modules/work/**",
"modules/continuity/**",
"cmd/creator-web/**",
"deploy/**",
"security/**"
]
},
"allowed_architecture_delta": {
"new_cross_module_dependencies": 0,
"new_canonical_owners": 0,
"new_privileges": 0,
"new_third_party_dependencies": 0
},
"contracts": {
"consume": ["contracts.identity@v1", "contracts.capability@v1", "contracts.commit@v1", "ADR-0005 fixed receipt protocol"],
"may_change": ["CapabilityBinding V1", "Commit prospective notification evidence V1"]
},
"semantic_owner": "Capability owns binding state; Commit owns immutable action evidence; receiver storage is external adapter receipt state",
"canonical_state": ["capability.notification_bindings", "future Commit envelope columns only; no action rows created by this task"],
"state_transitions": ["CapabilityBinding ACTIVE to RETIRED with monotonic revision", "receiver receipt absent to immutable present"],
"side_effects_and_privileges": ["receiver PUT writes only its isolated receipt directory", "receiver GET/list are read-only", "HTTPS adapter accepts only a validated fixed binding", "no credentials installed and no route exposed to Creator"],
"persistence_and_migrations": ["add Capability-owned binding table", "add backward-compatible notification evidence columns to Commit envelope", "receiver uses atomic create and immutable receipt files"],
"failure_retry_restart": ["same intent and digest returns the original receipt", "same intent with different digest is conflict", "timeout or malformed success is unknown", "GET reconciliation never writes", "restart rebuilds receiver truth from durable receipt files"],
"assumptions": ["The public TLS reverse proxy is configured only in a later deployment task", "secret references are identifiers and never secret bytes", "docker-test remains prohibited"],
"acceptance_scenarios": ["binding validation rejects HTTP, non-fixed paths, invalid revisions, unhealthy or retired state", "owner-local binding load cannot return a foreign principal binding", "receiver authorization separates writer and reader", "receiver survives restart and keeps one receipt per action intent", "executor disables redirects, bounds response size, verifies receipt identity and digest, and classifies ambiguous failures as unknown", "historical Commit rows remain readable after additive migration", "no Creator command can invoke the new executor in this task"],
"required_tests": ["unit", "receiver restart and conflict", "TLS executor and reconciliation", "PostgreSQL repository contract where available", "architecture preflight", "exact diff scope", "go test", "go vet", "CI"],
"terminal_gap_states": ["BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT", "BLOCKED_BY_MISSING_CAPABILITY"]
}
+54
View File
@@ -0,0 +1,54 @@
package domain
import (
"errors"
"net/url"
"strings"
"git.cin.su/m/Assistant/contracts/capability"
)
var ErrInvalidBinding = errors.New("invalid capability binding")
func ValidateBinding(binding capability.Binding) error {
if binding.ID == "" || binding.CapabilityID == "" || binding.OwnerPrincipalID == "" ||
binding.Revision < 1 || binding.Protocol != capability.ProtocolIdempotentReceiptV1 ||
binding.SecretRef == "" || binding.VerifiedAtUnixMillis <= 0 {
return ErrInvalidBinding
}
if binding.State != capability.BindingActive && binding.State != capability.BindingRetired {
return ErrInvalidBinding
}
if binding.Health != capability.HealthAvailable && binding.Health != capability.HealthDegraded && binding.Health != capability.HealthUnavailable {
return ErrInvalidBinding
}
origin, err := url.Parse(binding.HTTPSOrigin)
if err != nil || origin.Scheme != "https" || origin.Host == "" || origin.User != nil || origin.RawQuery != "" || origin.Fragment != "" || (origin.Path != "" && origin.Path != "/") {
return ErrInvalidBinding
}
if !fixedPath(binding.DeliveryPath) || !fixedPath(binding.ReceiptPath) || binding.DeliveryPath == binding.ReceiptPath {
return ErrInvalidBinding
}
return nil
}
func UsableBinding(binding capability.Binding) error {
if err := ValidateBinding(binding); err != nil || binding.State != capability.BindingActive || binding.Health != capability.HealthAvailable {
return ErrInvalidBinding
}
return nil
}
func RetireBinding(binding capability.Binding) (capability.Binding, error) {
if err := ValidateBinding(binding); err != nil || binding.State != capability.BindingActive {
return capability.Binding{}, ErrInvalidBinding
}
binding.State = capability.BindingRetired
binding.Revision++
return binding, nil
}
func fixedPath(value string) bool {
return strings.HasPrefix(value, "/") && value != "/" && !strings.HasSuffix(value, "/") &&
!strings.Contains(value, "//") && !strings.ContainsAny(value, "?#\\") && !strings.Contains(value, "..")
}
+45
View File
@@ -0,0 +1,45 @@
package domain
import (
"testing"
"git.cin.su/m/Assistant/contracts/capability"
"git.cin.su/m/Assistant/contracts/identity"
)
func validBinding() capability.Binding {
return capability.Binding{ID: "binding-1", CapabilityID: "prospective-notification-delivery", OwnerPrincipalID: identity.PrincipalID("owner-1"), Revision: 1, State: capability.BindingActive, HTTPSOrigin: "https://assistent.cin.su", DeliveryPath: "/delivery/v1/receipts", ReceiptPath: "/delivery/v1/receipts", Protocol: capability.ProtocolIdempotentReceiptV1, SecretRef: "notification-write-v1", VerifiedAtUnixMillis: 1, Health: capability.HealthAvailable}
}
func TestValidateBinding(t *testing.T) {
b := validBinding()
b.ReceiptPath = "/delivery/v1/lookup"
if err := ValidateBinding(b); err != nil {
t.Fatal(err)
}
for _, mutate := range []func(*capability.Binding){
func(v *capability.Binding) { v.HTTPSOrigin = "http://assistent.cin.su" },
func(v *capability.Binding) { v.DeliveryPath = "//evil" },
func(v *capability.Binding) { v.SecretRef = "" },
func(v *capability.Binding) { v.Revision = 0 },
func(v *capability.Binding) { v.Health = capability.HealthDegraded },
} {
candidate := b
mutate(&candidate)
if err := UsableBinding(candidate); err == nil {
t.Fatalf("expected rejection: %+v", candidate)
}
}
}
func TestRetireBindingIncrementsRevision(t *testing.T) {
b := validBinding()
b.ReceiptPath = "/delivery/v1/lookup"
retired, err := RetireBinding(b)
if err != nil {
t.Fatal(err)
}
if retired.State != capability.BindingRetired || retired.Revision != 2 {
t.Fatalf("unexpected retired binding: %+v", retired)
}
}
+4
View File
@@ -4,10 +4,14 @@ import (
"context"
"git.cin.su/m/Assistant/contracts/capability"
"git.cin.su/m/Assistant/contracts/identity"
)
type Repository interface {
SaveHealth(context.Context, capability.Health) error
LoadHealth(context.Context, capability.CapabilityID) (capability.Health, error)
RebuildHealthProjection(context.Context) (int64, error)
SaveBinding(context.Context, capability.Binding) error
LoadBinding(context.Context, capability.CapabilityBindingID, int64) (capability.Binding, error)
LoadActiveBinding(context.Context, capability.CapabilityID, identity.PrincipalID) (capability.Binding, error)
}