sow-tools/internal/nwsync/sink.go

package nwsync

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"io"
	"os"
	"path"
	"path/filepath"
	"strings"

	"git.westgate.pw/ShadowsOverWestgate/sow-tools/internal/depot"
)

// sink is where an emit or assemble run puts what it produces. The zone is the
// production sink; a local directory exists only as the conformance path, so
// upstream's output and ours can be diffed on a developer machine.
type sink interface {
	// putBlob stores one NWCompressedBuffer blob under its sha1 name and
	// returns the bytes stored, or 0 if a good copy was already there. Blob
	// names are content hashes, so an existing name is normally taken as
	// existing content — which is why body is a thunk: compression is the
	// expensive part of emit and a blob that is already stored must not pay
	// for it.
	//
	// verify stops trusting presence: the stored copy is read back, unwrapped
	// and hashed, and replaced when it is not what its name claims. Without it
	// an object written truncated, or written by an emitter since found broken,
	// is skipped by every later emit forever and no backfill can repair it.
	putBlob(sha1Hex string, verify bool, body func() []byte) (int64, error)
	// putIndex stores a NSYM manifest and its sidecar under key, which is
	// either an artifact-derived object key or a local path.
	putIndex(key string, manifest, sidecar []byte) error
	// getIndex reads back a NSYM manifest and its sidecar.
	getIndex(key string) (manifest, sidecar []byte, err error)
	// describe names the sink for messages.
	describe(key string) string
}

// dirSink writes a local NWSync repository tree.
type dirSink struct{ root string }

func (s dirSink) putBlob(sha1Hex string, verify bool, body func() []byte) (int64, error) {
	blob := blobPath(s.root, sha1Hex)
	if !verify {
		// Stat, not read: the common path must not pay to open every blob that
		// is already there.
		if _, err := os.Stat(blob); err == nil {
			return 0, nil
		}
	} else if stored, err := os.ReadFile(blob); err == nil && blobMatchesName(stored, sha1Hex) == nil {
		return 0, nil
	}
	if err := os.MkdirAll(filepath.Dir(blob), 0o755); err != nil {
		return 0, fmt.Errorf("create blob directory: %w", err)
	}
	data := body()
	if err := os.WriteFile(blob, data, 0o644); err != nil {
		return 0, fmt.Errorf("write blob: %w", err)
	}
	return int64(len(data)), nil
}

func (s dirSink) putIndex(key string, manifest, sidecar []byte) error {
	target := filepath.Join(s.root, filepath.FromSlash(key))
	if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
		return fmt.Errorf("create manifest directory: %w", err)
	}
	if err := os.WriteFile(target, manifest, 0o644); err != nil {
		return fmt.Errorf("write manifest: %w", err)
	}
	if err := os.WriteFile(target+".json", sidecar, 0o644); err != nil {
		return fmt.Errorf("write sidecar: %w", err)
	}
	return nil
}

func (s dirSink) getIndex(key string) ([]byte, []byte, error) {
	target := filepath.Join(s.root, filepath.FromSlash(key))
	manifest, err := os.ReadFile(target)
	if err != nil {
		return nil, nil, fmt.Errorf("read index: %w", err)
	}
	sidecar, err := os.ReadFile(target + ".json")
	if err != nil {
		return nil, nil, fmt.Errorf("read sidecar: %w", err)
	}
	return manifest, sidecar, nil
}

func (s dirSink) describe(key string) string {
	return filepath.Join(s.root, filepath.FromSlash(key))
}

// zoneSink uploads straight to the NWSync storage zone. Nothing bulky is ever
// written to the runner's disk: the working set is one resource at a time.
type zoneSink struct {
	store depot.KeyStore
	ctx   context.Context
	zone  string
}

func (s zoneSink) putBlob(sha1Hex string, verify bool, body func() []byte) (int64, error) {
	key := blobKey(sha1Hex)
	// A throttled probe must never be read as "missing, re-upload" or as
	// "present, skip", so only a confirmed Present skips the upload.
	state, _, err := s.store.ProbeKey(s.ctx, key)
	if err != nil {
		return 0, fmt.Errorf("probe blob %s: %w", sha1Hex, err)
	}
	if state == depot.Present {
		if !verify {
			return 0, nil
		}
		// The probe only proved the object exists. Read it back and hold it to
		// its own name.
		//
		// This reads the storage API rather than the pull zone: emit holds the
		// write credential, and a repair decision has to be made against the
		// copy it is about to overwrite, not against an edge cache of it. A
		// read that fails outright is a fault, not a verdict — treating it as
		// "bad, re-upload" would turn a throttled zone into a full backfill.
		stored, err := s.store.GetKey(s.ctx, key)
		if err != nil {
			return 0, fmt.Errorf("read back blob %s: %w", sha1Hex, err)
		}
		if blobMatchesName(stored, sha1Hex) == nil {
			return 0, nil
		}
	}
	data := body()
	if err := s.put(key, data); err != nil {
		return 0, fmt.Errorf("upload blob %s: %w", sha1Hex, err)
	}
	return int64(len(data)), nil
}

func (s zoneSink) putIndex(key string, manifest, sidecar []byte) error {
	// The manifest lands last: its presence is the publication marker, so it
	// must never appear before the blobs it names.
	if err := s.put(key+".json", sidecar); err != nil {
		return fmt.Errorf("upload sidecar %s: %w", key, err)
	}
	if err := s.put(key, manifest); err != nil {
		return fmt.Errorf("upload index %s: %w", key, err)
	}
	return nil
}

func (s zoneSink) getIndex(key string) ([]byte, []byte, error) {
	manifest, err := s.store.GetKey(s.ctx, key)
	if err != nil {
		return nil, nil, fmt.Errorf("read index: %w", err)
	}
	sidecar, err := s.store.GetKey(s.ctx, key+".json")
	if err != nil {
		return nil, nil, fmt.Errorf("read sidecar: %w", err)
	}
	return manifest, sidecar, nil
}

func (s zoneSink) describe(key string) string { return s.zone + "/" + key }

func (s zoneSink) put(key string, body []byte) error {
	sum := sha256.Sum256(body)
	return s.store.PutReader(s.ctx, key, bytes.NewReader(body), int64(len(body)), hex.EncodeToString(sum[:]))
}

// newZoneSink builds the upload sink from the environment. NWSync data lives
// in its own zone, separate from the asset depot, so it has its own zone and
// credential; only the host is shared, and Crucible has no default host.
func newZoneSink(ctx context.Context, getenv func(string) string) (sink, error) {
	cfg := depot.LoadConfig(getenv)
	cfg.StorageZone = getenv("NWSYNC_STORAGE_ZONE")
	cfg.WriteKey = getenv("NWSYNC_STORAGE_PASSWORD")
	cfg.ReadKey = cfg.WriteKey
	if cfg.StorageZone == "" {
		return nil, fmt.Errorf("NWSYNC_STORAGE_ZONE is unset (or pass --out DIR to write locally)")
	}
	if cfg.WriteKey == "" {
		return nil, fmt.Errorf("NWSYNC_STORAGE_PASSWORD is unset (or pass --out DIR to write locally)")
	}
	if cfg.StorageHost == "" {
		return nil, fmt.Errorf("BUNNY_STORAGE_HOST is unset")
	}
	store, err := depot.NewKeyStore(cfg)
	if err != nil {
		return nil, err
	}
	return zoneSink{store: store, ctx: ctx, zone: cfg.StorageZone}, nil
}

// indexKey is where an artifact's NSYM lives: beside the artifact itself, with
// the final extension replaced. emit and assemble must agree on this one rule,
// so it lives here and nowhere else.
//
// artifacts/haks/sha256/30/46/3046….hak -> artifacts/haks/sha256/30/46/3046….nsym
func indexKey(artifactKey string) (string, error) {
	extension := path.Ext(artifactKey)
	if extension == "" {
		return "", fmt.Errorf("artifact key %q has no extension", artifactKey)
	}
	return strings.TrimSuffix(artifactKey, extension) + ".nsym", nil
}

// resolveIndexKey is where emit writes an artifact's index and where assemble
// reads it from. On the zone that is beside the artifact; locally the indexes
// sit flat beside the data tree, so upstream's output and ours diff directly.
func resolveIndexKey(artifactKey, outDir string) (string, error) {
	key, err := indexKey(artifactKey)
	if err != nil {
		return "", err
	}
	if outDir != "" {
		return path.Base(key), nil
	}
	return key, nil
}

// checkArtifactKey fails closed when the key's embedded digest is not the
// digest of the bytes being emitted. Publishing an index under the wrong key
// silently pairs a manifest with the wrong artifact.
// artifact is hashed by streaming, so a multi-gigabyte hak is never resident.
func checkArtifactKey(artifactKey string, artifact io.Reader) error {
	base := path.Base(artifactKey)
	digest := strings.TrimSuffix(base, path.Ext(base))
	if len(digest) != 64 {
		return fmt.Errorf("artifact key %q does not name a sha256", artifactKey)
	}
	hash := sha256.New()
	if _, err := io.Copy(hash, artifact); err != nil {
		return fmt.Errorf("hash artifact: %w", err)
	}
	if got := hex.EncodeToString(hash.Sum(nil)); got != digest {
		return fmt.Errorf("artifact key %q names digest %s but the file hashes to %s", artifactKey, digest, got)
	}
	return nil
}