package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"

	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/provision"
)

// The cut's archive (moves spec §7.4 staged, §8.3, codex 19, ruling G14):
// how the stopped files directory's one sealed archive is identified,
// published create-only into this box's move-snapshots/ and staged into the
// lane's files/, and how a replay finishes exactly the identity an earlier
// pass journalled. The cut's phase loop is in movescut.go; this file is the
// archive's publication and its recovery, on their own so the recovery
// rules can be read in one place (Task 5 fix round 1, ruling 7).
//
// Source-disk headroom. The preparing round keeps the newest built archive
// on the source's data disk under <JournalDir>/archive/ for the life of
// the move, so the cut can publish the very object the target verified
// (G14). That costs up to archive.Cap (256 MiB) of the source's data disk
// while a move is in flight. A rebuild (the stopped files differ from the
// verified archive) is worse for the length of one build: the previous
// sealed archive is still there, buildArchive writes the compressed
// plaintext (archive.tar.zst) to a temp directory beside it, and then
// seals that into the new sealed archive before pruneArchives removes the
// previous one. With incompressible files all three approach the payload
// size, so the worst case is about three times archive.Cap plus the
// sealing overhead. The target's T6 does not price this source-side
// headroom; that is a Plan 6 follow-up.
// cutArchiveStep is the archive's identity in the op journal, written
// before any PUT so a replay publishes the same object (codex 19): the
// archive's name and digests, the two move-snapshots/ keys it is published
// under, and the manifest's bytes, so a replay that finds the sealed
// object published and the manifest not can write the manifest without
// the file (ruling 4). Reuse marks the preparing archive the target
// verified (G14): the seal names its staged versions and nothing is staged
// again.
type cutArchiveStep struct {
	builtArchive
	Reuse       bool            `json:"reuse,omitempty"`
	SealedKey   string          `json:"sealed_key"`
	ManifestKey string          `json:"manifest_key"`
	Manifest    json.RawMessage `json:"manifest"`
}

// errArchiveLost: the journalled identity can be finished by nobody. The
// sealed file is gone from this disk and the sealed object was never
// published, so there is nothing to publish it from, and a new identity
// is not allowed once one is journalled (ruling 4). The cut reverts.
var errArchiveLost = errors.New("the cut's archive is lost before it was published")

// cutArchive publishes the cut's one sealed archive (§8.3) and returns the
// seal's files: the sealed object and its manifest by version.
//
// Its identity is fixed before any PUT (sealing is randomized), in the op
// journal, and once journalled it is final: a replay finishes that
// identity or the cut reverts, and never reseals under a new name
// (ruling 4). The first run reuses the preparing archive when its
// plaintext digest is the stopped files' and its staged versions are pinned
// and its sealed file is still on this disk (G14); otherwise it builds and
// seals a new one. Then, from the journalled identity (publishCutArchive):
//   - the sealed file on disk: create-only into move-snapshots/ (an existing
//     object is this cut's earlier PUT), then staged into files/ unless it
//     is the reused archive, whose staged versions the ledger holds;
//   - no file: what move-snapshots/ holds decides, and only a listing that
//     answered decides it (an error is an error, never absence). The
//     sealed object there and the manifest not: the manifest is written
//     from the journalled bytes. Both there: staged into files/ by a
//     guarded server-side copy (or, reused, nothing to stage). The sealed
//     object absent: errArchiveLost, which is a revert.
//
// why is a recorded reason the stopped files cannot be archived.
func (c *cutRun) cutArchive(ctx context.Context, plain string) (builtArchive, []moves.SealedObject, string, error) {
	cfg := c.cfg()
	host, id := c.rec.Host, c.g.ID
	own, vs, err := c.ownStores(ctx)
	if err != nil {
		return builtArchive{}, nil, "", err
	}
	ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, moves.LedgerKey(host, id))
	if err != nil {
		return builtArchive{}, nil, "", err
	}
	j, err := c.journal()
	if err != nil {
		return builtArchive{}, nil, "", err
	}
	adir := archiveDir(cfg.DataDir, host, id)
	snap := moves.SnapshotsPrefix(host, id)
	var step cutArchiveStep
	if raw, ok := j.Steps[cutStepArchive]; ok {
		// Journalled: final. An entry that cannot be read, or names another
		// plaintext than the record's, is a corrupt journal, and starting
		// over on it could publish a second identity: an error.
		if err := json.Unmarshal([]byte(raw), &step); err != nil || step.Name != archiveName(step.Sealed) || step.Plain != plain ||
			step.SealedKey != snap+step.Name+".sealed" || step.ManifestKey != snap+step.Name+".json" || len(step.Manifest) == 0 {
			return builtArchive{}, nil, "", fmt.Errorf("the journalled archive identity of %s is unusable (%v)", id, err)
		}
		step.dir = adir
	} else {
		reuse := ledger.ArchivePlain == plain && ledger.ArchiveVersions != nil && ledger.ArchiveVersions.Pinned() && ledger.ArchiveSealed != ""
		if reuse {
			kept := builtArchive{Name: archiveName(ledger.ArchiveSealed), Sealed: ledger.ArchiveSealed, Plain: plain, dir: adir}
			keptThere, err := sealedOnDisk(kept)
			if err != nil {
				return builtArchive{}, nil, "", err
			}
			// The kept manifest is needed here, before anything is
			// journalled: without it the verified archive is simply not
			// reused, and a new one is built.
			if m, ok := keptManifest(kept); ok && keptThere {
				kept.Bytes, kept.Entries = m.Bytes, len(m.Entries)
				if st, err := os.Stat(kept.sealedPath()); err == nil {
					kept.SealedSize = st.Size()
				}
				step = cutArchiveStep{builtArchive: kept, Reuse: true}
			} else {
				reuse = false
			}
		}
		if !reuse {
			a, why, retry, err := c.w.buildArchive(c.rec)
			if err != nil || why != "" {
				return builtArchive{}, nil, why, err
			}
			if retry {
				return builtArchive{}, nil, "", errors.New("a stopped file changed while it was archived")
			}
			if a.Plain != plain {
				return builtArchive{}, nil, "", fmt.Errorf("the stopped files' digest is %s, recorded %s", a.Plain, plain)
			}
			step = cutArchiveStep{builtArchive: a}
		}
		manifest, err := os.ReadFile(step.manifestPath())
		if err != nil {
			return builtArchive{}, nil, "", err
		}
		step.SealedKey, step.ManifestKey, step.Manifest = snap+step.Name+".sealed", snap+step.Name+".json", manifest
		body, err := json.Marshal(step)
		if err != nil {
			return builtArchive{}, nil, "", err
		}
		j.Mark(cutStepArchive, string(body))
		if err := j.Commit(cfg.DataDir); err != nil {
			return builtArchive{}, nil, "", err
		}
		if err := c.fault("archive-journaled"); err != nil {
			return builtArchive{}, nil, "", err
		}
	}
	a := step.builtArchive
	files, err := c.publishCutArchive(ctx, own, vs, step, ledger)
	if errors.Is(err, errArchiveLost) {
		log.Printf("moves: %s: %v; the cut reverts", host, err)
		return builtArchive{}, nil, provision.RevertArchiveRefused, nil
	}
	return a, files, "", err
}

// publishCutArchive is cutArchive's PUTs from a journalled identity. The
// manifest it stages is the journalled one: the file on disk, when there
// is one, was written by the same build, and the journal is what a replay
// without the file has.
func (c *cutRun) publishCutArchive(ctx context.Context, own lease.MirrorStore, vs lease.VersionedStore, step cutArchiveStep, ledger moveLedger) ([]moves.SealedObject, error) {
	cfg := c.cfg()
	host, id := c.rec.Host, c.g.ID
	a := step.builtArchive
	snap := moves.SnapshotsPrefix(host, id)
	sealedSnap, manifestSnap := step.SealedKey, step.ManifestKey
	manifest := []byte(step.Manifest)
	onDisk, err := sealedOnDisk(a)
	if err != nil {
		return nil, err
	}
	if onDisk {
		fp, ok := vs.(lease.FilePutIfAbsent)
		if !ok {
			return nil, errors.New("this box cannot write the archive create-only")
		}
		if _, err := fp.PutFileIfAbsent(ctx, cfg.OwnBucket, sealedSnap, a.sealedPath()); err != nil && !errors.Is(err, lease.ErrExists) {
			return nil, fmt.Errorf("snapshot the sealed archive: %w", err)
		}
		if err := c.fault("archive-sealed-snapshot"); err != nil {
			return nil, err
		}
		if _, err := vs.PutIfAbsent(ctx, cfg.OwnBucket, manifestSnap, manifest); err != nil && !errors.Is(err, lease.ErrExists) {
			return nil, fmt.Errorf("snapshot the archive manifest: %w", err)
		}
		if err := c.fault("archive-own-put"); err != nil {
			return nil, err
		}
	} else {
		// No file (the read succeeded and found none, or a file that is not
		// this archive): only what was published under the identity can
		// finish it. A listing that failed decides nothing (ruling 4). The
		// loss is established only here, by reads that answered.
		sealedThere, _, err := c.snapshotState(ctx, vs, sealedSnap)
		if err != nil {
			return nil, err
		}
		if !sealedThere {
			return nil, fmt.Errorf("%w: %s is not on this disk and not in %s", errArchiveLost, a.Name, snap)
		}
		manifestThere, _, err := c.snapshotState(ctx, vs, manifestSnap)
		if err != nil {
			return nil, err
		}
		if !manifestThere {
			// A crash between the two PUTs: the manifest from the journal.
			if _, err := vs.PutIfAbsent(ctx, cfg.OwnBucket, manifestSnap, manifest); err != nil && !errors.Is(err, lease.ErrExists) {
				return nil, fmt.Errorf("snapshot the archive manifest from the journal: %w", err)
			}
			log.Printf("moves: %s: the manifest of %s was written from the journalled identity", host, a.Name)
		}
	}
	if step.Reuse {
		ref := ledger.ArchiveVersions
		if ref == nil || !ref.Pinned() || ledger.ArchiveSealed != a.Sealed {
			return nil, fmt.Errorf("the ledger no longer names the archive %s", a.Name)
		}
		return []moves.SealedObject{
			{Key: ref.SealedObjectKey, VersionID: ref.SealedVersion, Size: a.SealedSize},
			{Key: ref.ManifestKey, VersionID: ref.ManifestVersion, Size: int64(len(manifest))},
		}, nil
	}
	prefix := moves.FilesPrefix(host, id)
	sealedKey, manifestKey := prefix+a.Name+".sealed", prefix+a.Name+".json"
	var sv, mv string
	var msize int64
	if onDisk {
		if sv, err = c.gm.PutFileVersioned(ctx, c.gm.dst, sealedKey, a.sealedPath(), a.SealedSize, digestHex(a.Sealed)); err != nil {
			return nil, fmt.Errorf("stage the sealed archive: %w", err)
		}
		if err := c.fault("archive-first-copy"); err != nil {
			return nil, err
		}
		if mv, err = c.gm.PutVersioned(ctx, c.gm.dst, manifestKey, manifest); err != nil {
			return nil, fmt.Errorf("stage the archive manifest: %w", err)
		}
		msize = int64(len(manifest))
	} else {
		// The file is gone, the snapshot is there: a server-side copy from
		// it, guarded, as the replica lane copies.
		objs, err := own.List(ctx, cfg.OwnBucket, snap)
		if err != nil {
			return nil, fmt.Errorf("list the snapshots: %w", err)
		}
		so, sok := objs[sealedSnap]
		mo, mok := objs[manifestSnap]
		if !sok || !mok {
			return nil, fmt.Errorf("the snapshot of %s is incomplete", a.Name)
		}
		if _, sv, err = c.gm.CopyVersioned(ctx, cfg.OwnBucket, sealedSnap, so.ETag, c.gm.dst, sealedKey); err != nil {
			return nil, fmt.Errorf("stage the sealed archive: %w", err)
		}
		if err := c.fault("archive-first-copy"); err != nil {
			return nil, err
		}
		if _, mv, err = c.gm.CopyVersioned(ctx, cfg.OwnBucket, manifestSnap, mo.ETag, c.gm.dst, manifestKey); err != nil {
			return nil, fmt.Errorf("stage the archive manifest: %w", err)
		}
		a.SealedSize, msize = so.Size, mo.Size
	}
	if err := c.fault("archive-second-copy"); err != nil {
		return nil, err
	}
	if !lease.UsableVersionID(sv) || !lease.UsableVersionID(mv) {
		return nil, fmt.Errorf("the staged archive has no usable version ids (%q, %q)", sv, mv)
	}
	log.Printf("moves: %s: staged the cut's archive %s", host, a.Name)
	return []moves.SealedObject{{Key: sealedKey, VersionID: sv, Size: a.SealedSize}, {Key: manifestKey, VersionID: mv, Size: msize}}, nil
}

// sealedOnDisk reports whether the archive's sealed file is on this disk
// and hashes to its name. The sealed file alone is what a publication
// needs: the manifest's bytes are journalled with the identity, so losing
// only the local manifest loses nothing (fix round 2). Absence, and a file
// whose digest is not the name, are false with no error; a read that
// fails is an error, never absence, since a loss is declared on this
// answer and a read error establishes none.
func sealedOnDisk(a builtArchive) (bool, error) {
	sum, err := fileSHA256(a.sealedPath())
	if errors.Is(err, os.ErrNotExist) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("read %s: %w", a.sealedPath(), err)
	}
	return sum == a.Sealed, nil
}

// keptManifest reads the manifest of an archive kept on disk; it must
// describe that archive.
func keptManifest(a builtArchive) (archive.Manifest, bool) {
	var m archive.Manifest
	body, err := os.ReadFile(a.manifestPath())
	if err != nil || json.Unmarshal(body, &m) != nil || m.Sealed != a.Sealed || m.Plain != a.Plain {
		return m, false
	}
	return m, true
}

// snapshotState reports whether key holds a current object (not a delete
// marker) in this box's bucket, and its size. A listing that fails is an
// error, never absence (ruling 4): absence lets a replay decide the
// identity was never published.
func (c *cutRun) snapshotState(ctx context.Context, vs lease.VersionedStore, key string) (present bool, size int64, err error) {
	vers, err := vs.ListVersions(ctx, c.cfg().OwnBucket, key)
	if err != nil {
		return false, 0, fmt.Errorf("list %s: %w", key, err)
	}
	for _, v := range vers {
		if v.Key == key && v.IsLatest && !v.DeleteMarker {
			return true, v.Size, nil
		}
	}
	return false, 0, nil
}

// sealNamesArchive reports whether the seal's files are the archive with
// the sealed digest given.
func sealNamesArchive(seal moves.Seal, sealed string) bool {
	if sealed == "" {
		return false
	}
	name := archiveName(sealed)
	n := 0
	for _, o := range seal.Files {
		base := filepath.Base(o.Key)
		if base == name+".sealed" || base == name+".json" {
			n++
		}
	}
	return n == 2 && len(seal.Files) == 2
}