nwsync: upload sink, key-addressed CLI, fail-closed publication marker (#73)

Builds the code half of #60, and closes the CLI gap the #53 sweep found: PR #71 shipped #53's original surface rather than the one #56, #62 and #65 settled.

## What lands

**`depot.KeyStore`** — `ProbeKey` / `PutReader` / `GetKey`, addressing the zone by object key rather than by depot sha. NWSync cannot use the sha-addressed path: a blob is named after the sha1 of its *uncompressed* bytes while the body uploaded is the compressed form, and Bunny's `Checksum` header is sha256 of the body. Per #55 this reuses `internal/depot`'s `httpBackend` — same IPv4-pinned transport, same retry, same tri-state probe — and the sha-addressed `Backend` is now rewritten on top of it. No second HTTP client, no per-instance hash-function fields: the caller passes the key and the checksum, which turned out simpler than #55 expected.

**A sink in `internal/nwsync`** — the zone by default, a local tree under `--out DIR` as the conformance path. Blobs upload as they are produced and the index lands last, so the presence of an index is the publication marker. A blob already in the zone is skipped via #55's probe *without* paying for compression (the body is a thunk) — which matters for the backfill, where compression is the expensive part.

**The settled CLI**

```
nwsync emit     [--as NAME] [--out DIR] <artifact-key> <file>
nwsync assemble --group-id N [--tlk-key KEY] [--out DIR] <artifact-key>...
```

Artifact keys are depot keys; an index lives beside its artifact with the extension replaced (#62), derived in exactly one place so `emit` and `assemble` cannot disagree. Flags may now follow positionals — Go's `flag` stops at the first non-flag argument, which cost a run during #59.

**Fail-closed in two places** — an artifact key whose embedded digest does not match the file is refused (publishing an index under the wrong key silently pairs a manifest with the wrong artifact), and `assemble` refuses an artifact with no index rather than publishing a manifest missing a hak.

## Checks

`make check` green. Six new tests run against a Bunny-shaped `httptest` zone that verifies the `Checksum` header the way Bunny does: blobs-then-index ordering, skip-if-present, no index after a failed upload, key/file mismatch, and assemble reading indexes back out of the zone.

Conformance re-run through the new CLI against upstream `nwn_nwsync_write` 2.1.2 over `sow_vfxs_01.hak` (#59's oracle): the manifest is still **byte-identical**.

## Not in this PR

- **Live upload against the real zone.** The nwsync zone and its credential are #61, still open. Everything here is proven against a fake zone only.
- **Consumer wiring** — #65, in the three producer repos.
- **The mid-hak failure *policy*.** The mechanism is here (fail closed, orphan blobs left, re-run resumes); whether a module release may proceed when an emit failed is a human call, still open on #60.

🤖 Generated with [Claude Code](https://claude.com/claude-code)Reviewed-on: #73

Co-authored-by: vickydotbat <vickydotbat@tutamail.com>

internal/nwsync/sink.go

@@ -0,0 +1,207 @@
+package nwsync
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"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 the blob was already there. Blob names
+	// are content hashes, so an existing name is 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.
+	putBlob(sha1Hex string, 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, body func() []byte) (int64, error) {
+	blob := blobPath(s.root, sha1Hex)
+	if _, err := os.Stat(blob); err == 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, body func() []byte) (int64, error) {
+	key := path.Join("data", "sha1", sha1Hex[0:2], sha1Hex[2:4], 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 {
+		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.
+func checkArtifactKey(artifactKey string, artifact []byte) 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)
+	}
+	sum := sha256.Sum256(artifact)
+	if got := hex.EncodeToString(sum[:]); got != digest {
+		return fmt.Errorf("artifact key %q names digest %s but the file hashes to %s", artifactKey, digest, got)
+	}
+	return nil
+}