niko_trust/internal/server/store.go

// Package server is the trust relay: it stores signed TCE objects keyed by
// their content address and serves them over the JSON transport. It is a dumb
// store, not an authority.
//
// Invariants enforced here:
//
//   - INV-1: this package never imports internal/identity/signer and never
//     holds a signing key. It can store and serve, never forge.
//   - INV-3: the only identifiers are cryptographic (public keys, object
//     hashes). There are no database row ids in any stored or served object.
//   - INV-5: the relay answers "who said what", never "is this allowed". It
//     strict-decodes and content-addresses objects but does not apply trust
//     policy; consumers verify signatures locally.
package server

import (
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	"git.n1ko.dev/Niko/niko_trust/internal/protocol"
	"git.n1ko.dev/Niko/niko_trust/internal/tce"
	"git.n1ko.dev/Niko/niko_trust/internal/transport"
)

// Store is a content-addressed, in-memory store of signed objects.
//
// It strict-decodes every object on arrival, recomputes its object ID from the
// TCE bytes (rejecting any envelope whose supplied ID disagrees), and indexes
// it for query. It never verifies signatures: that is the consumer's job.
type Store struct {
	mu sync.RWMutex

	// dir is the optional on-disk location. When empty the store is purely
	// in-memory.
	dir string

	byID        map[string]*transport.Envelope
	byType      map[string]map[string]struct{} // type -> id set
	bySubject   map[string]map[string]struct{} // claim subject address
	byRecipient map[string]map[string]struct{} // request recipient address
	byClaimID   map[string]map[string]struct{} // revocation -> target claim id
	byRequest   map[string]map[string]struct{} // response -> request hash

	// answeredRequests records request hashes that already have a stored
	// response, to enforce the one-response-per-request rule.
	answeredRequests map[string]struct{}

	// confirmedRotations records rotation request ids that already have a
	// stored confirm: competing consents make a link unusable rather than
	// letting anyone pick a winner (docs/ROTATION.md).
	confirmedRotations map[string]struct{}

	// maxPerSubject bounds how many claims a single subject may have in the
	// store, to keep the in-memory indexes from growing without bound under a
	// hostile or buggy publisher.
	maxPerSubject int
}

// NewStore returns a store. If dir is non-empty, existing objects are loaded
// from disk and every subsequent Put is persisted there. maxPerSubject is the
// per-subject claim cap.
func NewStore(dir string, maxPerSubject int) *Store {
	s := &Store{
		dir:         dir,
		byID:        make(map[string]*transport.Envelope),
		byType:      make(map[string]map[string]struct{}),
		bySubject:   make(map[string]map[string]struct{}),
		byRecipient: make(map[string]map[string]struct{}),
		byClaimID:   make(map[string]map[string]struct{}),
		byRequest:   make(map[string]map[string]struct{}),

		answeredRequests:   make(map[string]struct{}),
		confirmedRotations: make(map[string]struct{}),
		maxPerSubject:      maxPerSubject,
	}
	if dir != "" {
		s.load()
	}
	return s
}

// Put stores one object from its raw TCE bytes and signature. It returns the
// authoritative object ID and whether this call created the entry (an
// idempotent replay returns created=false). The supplied objectID, if any,
// must match the recomputed one.
func (s *Store) Put(tceBytes, sig []byte, suppliedID string) (string, bool, error) {
	if len(tceBytes) > tce.MaxClaimTCE*2 {
		return "", false, fmt.Errorf("server: object too large")
	}
	_, obj, err := transport.DecodeObject(tceBytes)
	if err != nil {
		return "", false, fmt.Errorf("server: rejected: %w", err)
	}
	id := tce.ComputeID(tceBytes).String()
	if suppliedID != "" && suppliedID != id {
		return "", false, fmt.Errorf("server: object_id %s does not match recomputed %s", suppliedID, id)
	}

	s.mu.Lock()
	defer s.mu.Unlock()
	if _, exists := s.byID[id]; exists {
		return id, false, nil // idempotent
	}
	if claim, ok := obj.(*protocol.Claim); ok {
		sub := transport.AddrOf(claim.Subject)
		if len(s.bySubject[sub]) >= s.maxPerSubject {
			return "", false, fmt.Errorf("server: subject %s quota exceeded", sub)
		}
	}
	if resp, ok := obj.(*protocol.ApprovalResponse); ok {
		rh := resp.RequestHash.String()
		if _, done := s.answeredRequests[rh]; done {
			return "", false, fmt.Errorf("server: request %s already answered", rh)
		}
	}
	if conf, ok := obj.(*protocol.KeyRotationConfirm); ok {
		rh := conf.RotationHash.String()
		if _, done := s.confirmedRotations[rh]; done {
			return "", false, fmt.Errorf("server: rotation %s already confirmed", rh)
		}
	}
	s.addLocked(tceBytes, sig, id, obj)
	if s.dir != "" {
		if err := s.writeFile(id, tceBytes, sig); err != nil {
			return "", false, fmt.Errorf("server: persist: %w", err)
		}
	}
	return id, true, nil
}

// addLocked inserts an already-validated object into the in-memory indexes. The
// caller must hold s.mu.
func (s *Store) addLocked(tceBytes, sig []byte, id string, obj any) {
	s.byID[id] = &transport.Envelope{
		TCE:       append([]byte(nil), tceBytes...),
		Signature: append([]byte(nil), sig...),
	}
	typ := transport.ObjectTypeName(obj)
	addKey(s.byType, typ, id)

	switch o := obj.(type) {
	case *protocol.Claim:
		addKey(s.bySubject, transport.AddrOf(o.Subject), id)
	case *protocol.ApprovalRequest:
		addKey(s.byRecipient, transport.AddrOf(o.Recipient), id)
	case *protocol.Revocation:
		addKey(s.byClaimID, o.ClaimID.String(), id)
	case *protocol.ApprovalResponse:
		addKey(s.byRequest, o.RequestHash.String(), id)
		s.answeredRequests[o.RequestHash.String()] = struct{}{}
	case *protocol.KeyRotationConfirm:
		s.confirmedRotations[o.RotationHash.String()] = struct{}{}
	}
}

// writeFile atomically persists an envelope to dir/<id>.json.
func (s *Store) writeFile(id string, tceBytes, sig []byte) error {
	if err := os.MkdirAll(s.dir, 0o700); err != nil {
		return err
	}
	tmp := filepath.Join(s.dir, id+".tmp")
	final := filepath.Join(s.dir, id+".json")
	f, err := os.OpenFile(tmp, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600)
	if err != nil {
		return err
	}
	enc := json.NewEncoder(f)
	if err := enc.Encode(struct {
		TCE       []byte `json:"tce"`
		Signature []byte `json:"signature"`
		ObjectID  string `json:"object_id"`
	}{tceBytes, sig, id}); err != nil {
		f.Close()
		return err
	}
	if err := f.Close(); err != nil {
		return err
	}
	return os.Rename(tmp, final)
}

// load replays persisted envelopes from disk into the in-memory indexes. A
// corrupt or integrity-failing file is skipped.
func (s *Store) load() {
	entries, err := os.ReadDir(s.dir)
	if err != nil {
		return // no directory yet: start empty
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, e := range entries {
		if e.IsDir() || !strings.HasSuffix(e.Name(), ".json") {
			continue
		}
		raw, err := os.ReadFile(filepath.Join(s.dir, e.Name()))
		if err != nil {
			continue
		}
		var enc struct {
			TCE       []byte `json:"tce"`
			Signature []byte `json:"signature"`
			ObjectID  string `json:"object_id"`
		}
		if err := json.Unmarshal(raw, &enc); err != nil {
			continue
		}
		_, obj, err := transport.DecodeObject(enc.TCE)
		if err != nil {
			continue
		}
		id := tce.ComputeID(enc.TCE).String()
		if enc.ObjectID != id {
			continue // integrity check
		}
		if _, exists := s.byID[id]; exists {
			continue
		}
		s.addLocked(enc.TCE, enc.Signature, id, obj)
	}
}

// Get returns the stored envelope with its authoritative object_id and decoded
// `object` view filled in.
func (s *Store) Get(id string) (*transport.Envelope, error) {
	s.mu.RLock()
	env, ok := s.byID[id]
	s.mu.RUnlock()
	if !ok {
		return nil, fmt.Errorf("server: object %s not found", id)
	}
	return decorate(env)
}

// ClaimsBySubject returns all stored claims about the given subject address.
func (s *Store) ClaimsBySubject(addr string) ([]*transport.Envelope, error) {
	return s.queryIndex(s.bySubject, addr)
}

// RequestsByRecipient returns all stored approval requests for a recipient.
func (s *Store) RequestsByRecipient(addr string) ([]*transport.Envelope, error) {
	return s.queryIndex(s.byRecipient, addr)
}

// ResponsesForRequest returns all stored responses to a request hash.
func (s *Store) ResponsesForRequest(reqHash string) ([]*transport.Envelope, error) {
	return s.queryIndex(s.byRequest, reqHash)
}

// RevocationsForClaim returns all stored revocations targeting a claim object
// ID.
func (s *Store) RevocationsForClaim(claimID string) ([]*transport.Envelope, error) {
	return s.queryIndex(s.byClaimID, claimID)
}

func (s *Store) queryIndex(idx map[string]map[string]struct{}, key string) ([]*transport.Envelope, error) {
	s.mu.RLock()
	ids := idx[key]
	out := make([]*transport.Envelope, 0, len(ids))
	for id := range ids {
		if env, ok := s.byID[id]; ok {
			out = append(out, env)
		}
	}
	s.mu.RUnlock()
	// Lexicographic order over content addresses is deterministic, which
	// makes offset pagination meaningful and `after` cursors possible.
	sort.Slice(out, func(i, j int) bool {
		return tce.ComputeID(out[i].TCE).String() < tce.ComputeID(out[j].TCE).String()
	})
	for i, env := range out {
		d, err := decorate(env)
		if err != nil {
			return nil, err
		}
		out[i] = d
	}
	return out, nil
}

// GetMany returns decorated envelopes for the ids that exist, keyed by id.
// Missing ids are simply absent from the result.
func (s *Store) GetMany(ids []string) (map[string]*transport.Envelope, error) {
	s.mu.RLock()
	found := make([]*transport.Envelope, 0, len(ids))
	want := make(map[string]struct{}, len(ids))
	for _, id := range ids {
		want[id] = struct{}{}
		if env, ok := s.byID[id]; ok {
			found = append(found, env)
		}
	}
	s.mu.RUnlock()
	out := make(map[string]*transport.Envelope, len(found))
	for _, env := range found {
		id := tce.ComputeID(env.TCE).String()
		d, err := decorate(env)
		if err != nil {
			return nil, err
		}
		out[id] = d
	}
	return out, nil
}

// decorate fills the object_id and object view of a stored envelope.
func decorate(env *transport.Envelope) (*transport.Envelope, error) {
	id := tce.ComputeID(env.TCE).String()
	_, view, err := transport.BuildView(env.TCE)
	if err != nil {
		return nil, err
	}
	return &transport.Envelope{
		TCE:       env.TCE,
		Signature: env.Signature,
		ObjectID:  id,
		Object:    view,
	}, nil
}

func addKey(m map[string]map[string]struct{}, key, id string) {
	set := m[key]
	if set == nil {
		set = make(map[string]struct{})
		m[key] = set
	}
	set[id] = struct{}{}
}

// trieInsertLocked adds one object ID to the commitment. The caller holds

// now is overridable in tests.
var now = time.Now