86 lines
2.8 KiB
Go
86 lines
2.8 KiB
Go
|
// Package sink persists verified payloads to local storage. It is the
|
||
|
|
// trivial "write a file and be done" tail of the Phase-1 trust chain:
|
||
|
|
// each delivery writes `<DataDir>/<decision_id>.json` atomically via
|
||
|
|
// `.tmp` + rename so a crash cannot leave a half-written file.
|
||
|
|
package sink
|
||
|
|
|
||
|
|
import (
|
||
|
|
"encoding/json"
|
||
|
|
"errors"
|
||
|
|
"fmt"
|
||
|
|
"os"
|
||
|
|
"path/filepath"
|
||
|
|
"regexp"
|
||
|
|
)
|
||
|
|
|
||
|
|
// decisionIDPattern is deliberately strict so decision_id can safely be
|
||
|
|
// pasted into a filesystem path. Approval- and verification-services
|
||
|
|
// generate decision_id values from UUIDs / sha256 digests, so this covers
|
||
|
|
// their output and nothing dangerous.
|
||
|
|
var decisionIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$`)
|
||
|
|
|
||
|
|
// Store is a filesystem-backed payload store. All writes land in DataDir.
|
||
|
|
type Store struct {
|
||
|
|
DataDir string
|
||
|
|
}
|
||
|
|
|
||
|
|
// WriteResult is the record of an atomic payload write.
|
||
|
|
type WriteResult struct {
|
||
|
|
Path string
|
||
|
|
}
|
||
|
|
|
||
|
|
// ErrInvalidDecisionID is returned when decision_id would escape or
|
||
|
|
// otherwise be unsafe on the filesystem.
|
||
|
|
var ErrInvalidDecisionID = errors.New("sink: decision_id is not filesystem-safe")
|
||
|
|
|
||
|
|
// Write atomically writes payload to <DataDir>/<decisionID>.json.
|
||
|
|
// It writes to a sibling `<decisionID>.json.tmp` first and renames.
|
||
|
|
// Repeated calls with the same (decisionID) replace the file; replay
|
||
|
|
// protection is enforced by the verification-service per the Phase-1
|
||
|
|
// contract.
|
||
|
|
func (s Store) Write(decisionID string, payload any) (WriteResult, error) {
|
||
|
|
if s.DataDir == "" {
|
||
|
|
return WriteResult{}, errors.New("sink.Store: DataDir is required")
|
||
|
|
}
|
||
|
|
if !decisionIDPattern.MatchString(decisionID) {
|
||
|
|
return WriteResult{}, fmt.Errorf("%w: %q", ErrInvalidDecisionID, decisionID)
|
||
|
|
}
|
||
|
|
if err := os.MkdirAll(s.DataDir, 0o755); err != nil {
|
||
|
|
return WriteResult{}, fmt.Errorf("sink: ensure data dir: %w", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
body, err := json.MarshalIndent(payload, "", " ")
|
||
|
|
if err != nil {
|
||
|
|
return WriteResult{}, fmt.Errorf("sink: marshal payload: %w", err)
|
||
|
|
}
|
||
|
|
body = append(body, '\n')
|
||
|
|
|
||
|
|
final := filepath.Join(s.DataDir, decisionID+".json")
|
||
|
|
tmp := final + ".tmp"
|
||
|
|
|
||
|
|
// Open with O_TRUNC so a crashed previous .tmp cannot taint the write.
|
||
|
|
f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o640)
|
||
|
|
if err != nil {
|
||
|
|
return WriteResult{}, fmt.Errorf("sink: open tmp: %w", err)
|
||
|
|
}
|
||
|
|
if _, err := f.Write(body); err != nil {
|
||
|
|
_ = f.Close()
|
||
|
|
_ = os.Remove(tmp)
|
||
|
|
return WriteResult{}, fmt.Errorf("sink: write tmp: %w", err)
|
||
|
|
}
|
||
|
|
if err := f.Sync(); err != nil {
|
||
|
|
_ = f.Close()
|
||
|
|
_ = os.Remove(tmp)
|
||
|
|
return WriteResult{}, fmt.Errorf("sink: fsync tmp: %w", err)
|
||
|
|
}
|
||
|
|
if err := f.Close(); err != nil {
|
||
|
|
_ = os.Remove(tmp)
|
||
|
|
return WriteResult{}, fmt.Errorf("sink: close tmp: %w", err)
|
||
|
|
}
|
||
|
|
if err := os.Rename(tmp, final); err != nil {
|
||
|
|
_ = os.Remove(tmp)
|
||
|
|
return WriteResult{}, fmt.Errorf("sink: rename to final: %w", err)
|
||
|
|
}
|
||
|
|
return WriteResult{Path: final}, nil
|
||
|
|
}
|