orca/internal/storage/replication.go

// Package storage implements the CLI-side Syncthing replication logic
// for per-namespace storage replication (REQ-081, R-005). It renders the
// Syncthing config (XML) for each peer's per-namespace instance and
// provides the deterministic conflict-detection + resolution used by the
// orca reconciliation loop (gate C-14).
//
// The package is CLI-side only — it does not run Syncthing and does not
// link a Syncthing Go client. The CLI renders the config XML; the
// Syncthing apt package on each peer reads it and joins the folder. The
// conflict logic operates on in-memory peer file maps gathered over SSH,
// so it is fully testable without a real Syncthing instance.
package storage

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/xml"
	"errors"
	"fmt"
	"sort"
	"strings"
)

// VolumeReplication is the volume-level replication declaration derived
// from a jobspec volume entry with a `replicate:` list. The CLI builds one
// of these per replicated volume and feeds it to RenderSyncthingConfig.
//
//   - Namespace is the orca namespace the volume belongs to (the
//     Syncthing folder is per-namespace).
//   - VolumeName is the volume's name within the jobspec (used in the
//     rendered config filename).
//   - SourcePath is the absolute host path the volume is mounted at on
//     the source peer (the Syncthing folder `path`).
//   - ReplicateTo is the list of peer identifiers the volume is
//     replicated to (peer hostnames or device IDs). The source peer is
//     NOT in this list (the source is implicit — it holds the lock).
//   - SyncMode is "sendreceive" (default) or "sendonly". Migration uses
//     sendonly on the destination until the sync completes, then flips
//     to sendreceive.
type VolumeReplication struct {
	Namespace   string
	VolumeName  string
	SourcePath  string
	ReplicateTo []string
	SyncMode    string
}

// SyncthingConfig is the rendered (in-memory) Syncthing config for a
// single peer's per-namespace folder. The XML emitter
// (RenderSyncthingXML) serializes this into the `config.xml` file.
//
//   - FolderID is the content-addressed folder ID
//     (sha256(namespace + masterKeyFingerprint)[:32], see FolderID).
//   - Path is the on-disk path the folder is rooted at (SourcePath).
//   - Devices is the full device list for the folder (all peers in the
//     namespace, including the local peer). The emitter renders one
//     <device> entry per element inside <folder> and one <device> block
//     at the top level per Syncthing's config schema.
type SyncthingConfig struct {
	FolderID string
	Path     string
	Devices  []SyncthingDevice
}

// SyncthingDevice is a single peer device entry in the rendered config.
//
//   - ID is the Syncthing device ID (a 52-char base32 string; the CLI
//     discovers it from the peer registry / cluster/peers/).
//   - Name is the human-readable peer name (the orca node hostname).
//   - Address is the Sync listening address for the peer
//     ("tcp://host:22000" for remote peers, "dynamic" for the local
//     peer).
type SyncthingDevice struct {
	ID      string
	Name    string
	Address string
}

// Conflict is a single detected file-level conflict across peers. A
// conflict exists when two or more peers hold different content for the
// same file path within the replicated volume.
//
//   - Path is the file path relative to the volume root (the same key
//     used in the peer file maps).
//   - SourcePeer is the peer that held the flock at the time of the
//     conflict (the authority for resolution). Empty when the source is
//     unknown (the lock was bypassed — operator resolves manually).
//   - Versions maps peer → content for every peer that holds a copy of
//     the file. Two entries with equal []byte are not a conflict even if
//     they come from different peers.
type Conflict struct {
	Path       string
	SourcePeer string
	Versions   map[string][]byte
}

// masterKeyFingerprintPlaceholder is the default master-key fingerprint
// used when the caller has not yet wired the real step-ca master key
// (P10). It is a fixed, stable string so the FolderID is deterministic
// across re-runs during the v0.9 development window. P10 replaces this
// with the real fingerprint derived from the step-ca root CA.
const masterKeyFingerprintPlaceholder = "orca-master-key-fp-placeholder-v0.9"

// FolderID returns the content-addressed Syncthing folder ID for the
// given namespace + master-key fingerprint (REQ-081). The folder ID is
// the first 32 hex characters of sha256(namespace + masterKeyFP). The
// fingerprint is optional — when empty, the v0.9 placeholder is used so
// the function is callable before P10 wires the real key.
//
// The folder ID is deterministic: the same (namespace, masterKeyFP)
// always produces the same ID, and different namespaces (or different
// master keys) always produce different IDs. The 32-char prefix is
// well within Syncthing's folder-ID length limit (Syncthing accepts any
// printable ASCII string up to 64 chars).
func FolderID(namespace string, masterKeyFP string) string {
	if strings.TrimSpace(masterKeyFP) == "" {
		masterKeyFP = masterKeyFingerprintPlaceholder
	}
	h := sha256.Sum256([]byte(namespace + masterKeyFP))
	return hex.EncodeToString(h[:16])
}

// RenderSyncthingConfig builds the in-memory SyncthingConfig for the
// given volume replication + peer device list. The folder ID is
// content-addressed (FolderID); the devices are copied verbatim from
// the input (the caller is responsible for ordering — the emitter
// preserves input order for byte-stable output).
//
// Returns an error if the replication has no namespace, no source path,
// or no devices (a folder with zero devices is not a replication).
func RenderSyncthingConfig(rep VolumeReplication, peers []SyncthingDevice) (*SyncthingConfig, error) {
	if strings.TrimSpace(rep.Namespace) == "" {
		return nil, errors.New("storage: replication namespace is empty")
	}
	if strings.TrimSpace(rep.SourcePath) == "" {
		return nil, errors.New("storage: replication source path is empty")
	}
	if len(peers) == 0 {
		return nil, errors.New("storage: replication has no peer devices")
	}
	mode := strings.TrimSpace(rep.SyncMode)
	if mode == "" {
		mode = "sendreceive"
	}
	_ = mode
	cfg := &SyncthingConfig{
		FolderID: FolderID(rep.Namespace, ""),
		Path:     rep.SourcePath,
		Devices:  append([]SyncthingDevice(nil), peers...),
	}
	return cfg, nil
}

// syncthingXMLFolder is the <folder> element in the rendered config.
type syncthingXMLFolder struct {
	XMLName     xml.Name             `xml:"folder"`
	ID          string               `xml:"id,attr"`
	Path        string               `xml:"path,attr"`
	Type        string               `xml:"type,attr"`
	IgnorePerms bool                 `xml:"ignorePerms,attr"`
	Devices     []syncthingXMLDevice `xml:"device"`
	FSync       bool                 `xml:"fsync"`
}

// syncthingXMLDevice is the <device> element (both inside <folder> and
// at the top level; the top-level form carries the address).
type syncthingXMLDevice struct {
	XMLName     xml.Name `xml:"device"`
	ID          string   `xml:"id,attr"`
	Name        string   `xml:"name,attr"`
	Compression string   `xml:"compression,attr,omitempty"`
	Address     string   `xml:"address,omitempty"`
}

// syncthingXMLOptions is the <options> element.
type syncthingXMLOptions struct {
	XMLName               xml.Name `xml:"options"`
	ListenAddress         string   `xml:"listenAddress"`
	GlobalAnnounceEnabled bool     `xml:"globalAnnounceEnabled"`
	LocalAnnounceEnabled  bool     `xml:"localAnnounceEnabled"`
	RelayingEnabled       bool     `xml:"relayingEnabled"`
	URAccepted            int      `xml:"urAccepted"`
}

// syncthingXMLGUI is the <gui> element (disabled — no GUI).
type syncthingXMLGUI struct {
	XMLName xml.Name `xml:"gui"`
	Enabled bool     `xml:"enabled,attr"`
}

// syncthingXMLConfig is the root <configuration> element.
type syncthingXMLConfig struct {
	XMLName xml.Name             `xml:"configuration"`
	Version int                  `xml:"version,attr"`
	GUI     syncthingXMLGUI      `xml:"gui"`
	Options syncthingXMLOptions  `xml:"options"`
	Folders []syncthingXMLFolder `xml:"folder"`
	Devices []syncthingXMLDevice `xml:"device"`
}

// RenderSyncthingXML serializes the SyncthingConfig into the
// `config.xml` file content. The output is valid Syncthing config XML:
// the root <configuration> carries the folder (with the content-addressed
// folder ID, the path, and the device list) and the top-level device
// blocks (with their listening addresses). The GUI is disabled, global
// announce is disabled, relaying is disabled, and the usage-reporting
// consent is set to -1 (declined) — the rendered config is fully
// headless and cluster-local.
//
// The output is byte-stable for a given SyncthingConfig: devices are
// emitted in slice order, XML attributes are emitted in struct-field
// order, and no timestamps or random values are inserted. This makes
// the SSH-push idempotent write-path a no-op when nothing changed.
//
// Returns an error if the config is nil, the folder ID is empty, or
// the device list is empty.
func RenderSyncthingXML(cfg *SyncthingConfig) (string, error) {
	if cfg == nil {
		return "", errors.New("storage: syncthing config is nil")
	}
	if strings.TrimSpace(cfg.FolderID) == "" {
		return "", errors.New("storage: syncthing folder ID is empty")
	}
	if len(cfg.Devices) == 0 {
		return "", errors.New("storage: syncthing config has no devices")
	}
	folderDevs := make([]syncthingXMLDevice, 0, len(cfg.Devices))
	topDevs := make([]syncthingXMLDevice, 0, len(cfg.Devices))
	for _, d := range cfg.Devices {
		folderDevs = append(folderDevs, syncthingXMLDevice{
			ID:   d.ID,
			Name: d.Name,
		})
		topDevs = append(topDevs, syncthingXMLDevice{
			ID:          d.ID,
			Name:        d.Name,
			Compression: "metadata",
			Address:     d.Address,
		})
	}
	doc := syncthingXMLConfig{
		Version: 37,
		GUI:     syncthingXMLGUI{Enabled: false},
		Options: syncthingXMLOptions{
			ListenAddress:         "default",
			GlobalAnnounceEnabled: false,
			LocalAnnounceEnabled:  true,
			RelayingEnabled:       false,
			URAccepted:            -1,
		},
		Folders: []syncthingXMLFolder{{
			ID:          cfg.FolderID,
			Path:        cfg.Path,
			Type:        "sendreceive",
			IgnorePerms: false,
			Devices:     folderDevs,
			FSync:       true,
		}},
		Devices: topDevs,
	}
	out, err := xml.MarshalIndent(doc, "", "  ")
	if err != nil {
		return "", fmt.Errorf("storage: marshal syncthing xml: %w", err)
	}
	return xml.Header + string(out) + "\n", nil
}

// DetectConflicts scans the peer file maps for conflicting versions of
// the same file (gate C-14). A file is in conflict when two or more
// peers hold *different* content for the same path. Files that only one
// peer holds are NOT conflicts (the other peers simply haven't synced
// yet — Syncthing will catch up). Files that all peers hold with equal
// content are NOT conflicts.
//
// The peerFiles map is peer → (path → content). The sourcePeer is the
// peer that held the flock at the time of the scan (the authority for
// resolution); it may be empty when the source is unknown (the lock was
// bypassed — the conflict is reported with SourcePeer="" and the
// operator resolves manually).
//
// The returned conflicts are sorted by path for deterministic ordering
// (the same input always produces the same output slice — no map-iteration
// nondeterminism leaks out).
func DetectConflicts(namespace string, peerFiles map[string]map[string][]byte) ([]Conflict, error) {
	_ = namespace
	if len(peerFiles) == 0 {
		return nil, nil
	}
	// path -> peer -> content
	byPath := make(map[string]map[string][]byte)
	for peer, files := range peerFiles {
		for path, content := range files {
			if byPath[path] == nil {
				byPath[path] = make(map[string][]byte)
			}
			byPath[path][peer] = append([]byte(nil), content...)
		}
	}
	paths := make([]string, 0, len(byPath))
	for p := range byPath {
		paths = append(paths, p)
	}
	sort.Strings(paths)
	var conflicts []Conflict
	for _, path := range paths {
		versions := byPath[path]
		if len(versions) < 2 {
			continue
		}
		if !contentsDiffer(versions) {
			continue
		}
		c := Conflict{
			Path:     path,
			Versions: versions,
		}
		conflicts = append(conflicts, c)
	}
	return conflicts, nil
}

// contentsDiffer reports whether the peer→content map holds at least
// two distinct content values.
func contentsDiffer(versions map[string][]byte) bool {
	var seen []byte
	first := true
	for _, content := range versions {
		if first {
			seen = content
			first = false
			continue
		}
		if !bytesEqual(seen, content) {
			return true
		}
	}
	return false
}

// bytesEqual is a thin wrapper over bytes.Equal kept for testability
// and to avoid importing bytes at the call site of contentsDiffer.
func bytesEqual(a, b []byte) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

// ResolveConflict resolves a single conflict by picking the source
// peer's content (the peer that held the lock). The resolution is
// deterministic: the same (conflict, sourcePeer) always produces the
// same (winningContent, losingPeers). Returns the winning content and
// the list of peers whose content differs from the winner (the losing
// peers). The losingPeers list is sorted for deterministic ordering.
//
// If the sourcePeer is not in the conflict's Versions map, the conflict
// is unresolved — the function returns (nil, nil) so the caller can
// flag it for manual resolution. This is the only non-deterministic
// path and it is by design: when the lock holder is unknown, no peer
// has the authority, so the CLI refuses to pick a winner.
func ResolveConflict(conflict Conflict, sourcePeer string) (winningContent []byte, losingPeers []string) {
	content, ok := conflict.Versions[sourcePeer]
	if !ok {
		return nil, nil
	}
	winningContent = append([]byte(nil), content...)
	losing := make([]string, 0, len(conflict.Versions))
	for peer, c := range conflict.Versions {
		if peer == sourcePeer {
			continue
		}
		if !bytesEqual(c, winningContent) {
			losing = append(losing, peer)
		}
	}
	sort.Strings(losing)
	return winningContent, losing
}