releasekit/internal/delivery/remote_assets.go

package delivery

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"os"
	"sort"

	"go.maximumdirect.net/releasekit/internal/artifact"
	"go.maximumdirect.net/releasekit/internal/forge"
)

type AssetState string

const (
	AssetMatching     AssetState = "matching"
	AssetMissing      AssetState = "missing"
	AssetConflicting  AssetState = "conflicting"
	AssetUnverifiable AssetState = "unverifiable"
)

type AssetIdentity struct {
	Filename string
	SHA256   string
}

type AssetInspection struct {
	Expected artifact.ExpectedAsset
	State    AssetState
	Existing []forge.Asset
	Err      error
}

type Inspection struct {
	Assets    []AssetInspection
	Unrelated []forge.Asset
}

type UploadResult struct {
	State       State
	Matching    []AssetIdentity
	Uploaded    []AssetIdentity
	Pending     []AssetIdentity
	Conflicting []AssetIdentity
	Uncertain   []AssetIdentity
}

func inspectRemoteAssets(ctx context.Context, backend forge.Backend, releaseHandle any, expected []artifact.ExpectedAsset) (Inspection, error) {
	if err := artifact.ValidateExpectedAssets(expected); err != nil {
		return Inspection{}, &forge.Failure{Kind: forge.FailureInvalid, Err: err}
	}
	listed, err := backend.ListAssets(ctx, releaseHandle)
	if err != nil {
		inspection := Inspection{Assets: make([]AssetInspection, 0, len(expected))}
		for _, item := range expected {
			inspection.Assets = append(inspection.Assets, AssetInspection{Expected: item, State: AssetUnverifiable, Err: err})
		}
		return inspection, err
	}

	byName := make(map[string][]forge.Asset, len(listed))
	for _, item := range listed {
		byName[item.Filename] = append(byName[item.Filename], item)
	}
	inspection := Inspection{Assets: make([]AssetInspection, 0, len(expected))}
	declared := make(map[string]struct{}, len(expected))
	for _, item := range expected {
		declared[item.Filename] = struct{}{}
		matches := byName[item.Filename]
		observation := AssetInspection{Expected: item, Existing: matches}
		switch len(matches) {
		case 0:
			observation.State = AssetMissing
		case 1:
			digest, sizeMatches, err := remoteAssetDigest(ctx, backend, matches[0].Handle, item.Size)
			if err != nil {
				observation.State = AssetUnverifiable
				observation.Err = err
			} else if sizeMatches && digest == item.SHA256 {
				observation.State = AssetMatching
			} else {
				observation.State = AssetConflicting
			}
		default:
			observation.State = AssetConflicting
		}
		inspection.Assets = append(inspection.Assets, observation)
	}
	for _, item := range listed {
		if _, ok := declared[item.Filename]; !ok {
			inspection.Unrelated = append(inspection.Unrelated, item)
		}
	}
	return inspection, nil
}

func remoteAssetDigest(ctx context.Context, backend forge.Backend, handle any, expectedSize int64) (string, bool, error) {
	if err := ctx.Err(); err != nil {
		return "", false, err
	}
	stream, err := backend.OpenAsset(ctx, handle)
	if err != nil {
		return "", false, err
	}
	hasher := sha256.New()
	read, readErr := io.Copy(hasher, io.LimitReader(stream, expectedSize))
	excess := false
	if readErr == nil && read == expectedSize {
		var probe [1]byte
		for attempts := 0; ; attempts++ {
			var n int
			n, readErr = stream.Read(probe[:])
			excess = n != 0
			if readErr != nil || excess {
				if readErr == io.EOF {
					readErr = nil
				}
				break
			}
			if attempts == 99 {
				readErr = io.ErrNoProgress
				break
			}
		}
	}
	closeErr := stream.Close()
	if err := errors.Join(readErr, closeErr, ctx.Err()); err != nil {
		return "", false, &forge.Failure{Kind: forge.FailureUnverifiable, Err: err}
	}
	return hex.EncodeToString(hasher.Sum(nil)), read == expectedSize && !excess, nil
}

func uploadMissingAssets(ctx context.Context, backend forge.Backend, releaseHandle any, inspection Inspection) (result UploadResult, err error) {
	expected := make([]artifact.ExpectedAsset, len(inspection.Assets))
	for index, observation := range inspection.Assets {
		expected[index] = observation.Expected
	}
	if err := artifact.ValidateExpectedAssets(expected); err != nil {
		return UploadResult{State: Partial}, &forge.Failure{Kind: forge.FailureInvalid, Err: err}
	}
	result = UploadResult{State: Partial}
	preflightConflict := false
	preflightUnverifiable := false
	var inspectionErrors []error
	var missing []artifact.ExpectedAsset
	pending := make(map[string]struct{}, len(expected))
	// Keep the initial observation order and compact pending identities once,
	// including on early returns after partial or unresolved uploads.
	defer func() {
		remaining := result.Pending[:0]
		for _, item := range result.Pending {
			if _, ok := pending[item.Filename]; ok {
				remaining = append(remaining, item)
			}
		}
		result.Pending = remaining
	}()
	for _, observation := range inspection.Assets {
		item := identity(observation.Expected)
		switch observation.State {
		case AssetMatching:
			result.Matching = append(result.Matching, item)
		case AssetMissing:
			result.Pending = append(result.Pending, item)
			pending[observation.Expected.Filename] = struct{}{}
			missing = append(missing, observation.Expected)
		case AssetConflicting:
			preflightConflict = true
			result.Conflicting = append(result.Conflicting, item)
		case AssetUnverifiable:
			preflightUnverifiable = true
			result.Uncertain = append(result.Uncertain, item)
			if observation.Err != nil {
				inspectionErrors = append(inspectionErrors, fmt.Errorf("verify existing attachment %q: %w", item.Filename, observation.Err))
			}
		default:
			return result, &forge.Failure{Kind: forge.FailureInvalid, Err: fmt.Errorf("asset %q has no inspection result", observation.Expected.Filename)}
		}
	}
	if preflightConflict {
		result.State = Conflict
		return result, &forge.Failure{Kind: forge.FailureConflict, Err: errors.Join(errors.New("one or more existing attachments conflict"), errors.Join(inspectionErrors...))}
	}
	if preflightUnverifiable {
		result.State = NotAttempted
		return result, &forge.Failure{Kind: forge.FailureUnverifiable, Err: errors.Join(errors.New("one or more existing attachments could not be verified"), errors.Join(inspectionErrors...))}
	}
	if len(pending) == 0 {
		result.State = Complete
		return result, nil
	}

	orderMissing(missing)
	for _, item := range missing {
		content, err := assetContent(item)
		if err != nil {
			return result, &forge.Failure{Kind: forge.FailureInvalid, Err: fmt.Errorf("open local release asset %q: %w", item.Filename, err)}
		}
		evidence, uploadErr := backend.UploadAsset(ctx, releaseHandle, item.Filename, content)
		closeErr := content.Close()
		if uploadErr == nil {
			result.Uploaded = append(result.Uploaded, identity(item))
			delete(pending, item.Filename)
			continue
		}
		operationErr := errors.Join(uploadErr, closeErr)
		if evidence.Disposition != forge.WriteConflict && evidence.Disposition != forge.WriteUncertain {
			return result, operationErr
		}
		delete(pending, item.Filename)
		if err := reconcileAssetUpload(ctx, backend, releaseHandle, item, evidence, operationErr, &result); err != nil {
			return result, err
		}
	}
	if len(pending) == 0 {
		result.State = Complete
	}
	return result, nil
}

func reconcileAssetUpload(ctx context.Context, backend forge.Backend, releaseHandle any, item artifact.ExpectedAsset, evidence forge.WriteEvidence, uploadErr error, result *UploadResult) error {
	inspection, inspectErr := inspectRemoteAssets(ctx, backend, releaseHandle, []artifact.ExpectedAsset{item})
	if inspectErr != nil || len(inspection.Assets) != 1 {
		result.State = Unknown
		result.Uncertain = append(result.Uncertain, identity(item))
		return &forge.Failure{Kind: forge.FailureUnknown, Err: errors.Join(uploadErr, inspectErr)}
	}
	observation := inspection.Assets[0]
	switch observation.State {
	case AssetMatching:
		result.Matching = append(result.Matching, identity(item))
		return nil
	case AssetConflicting:
		result.State = Conflict
		result.Conflicting = append(result.Conflicting, identity(item))
		return &forge.Failure{Kind: forge.FailureConflict, Err: uploadErr}
	case AssetMissing:
		if evidence.Disposition == forge.WriteConflict {
			result.State = Conflict
			result.Conflicting = append(result.Conflicting, identity(item))
			return &forge.Failure{Kind: forge.FailureConflict, Err: uploadErr}
		}
		result.State = Unknown
		result.Uncertain = append(result.Uncertain, identity(item))
		return &forge.Failure{Kind: forge.FailureUnknown, Err: uploadErr}
	default:
		result.State = Unknown
		result.Uncertain = append(result.Uncertain, identity(item))
		return &forge.Failure{Kind: forge.FailureUnknown, Err: errors.Join(uploadErr, observation.Err)}
	}
}

func orderMissing(missing []artifact.ExpectedAsset) {
	sort.Slice(missing, func(i, j int) bool {
		if missing[i].Filename == artifact.ManifestFilename {
			return false
		}
		if missing[j].Filename == artifact.ManifestFilename {
			return true
		}
		return missing[i].Filename < missing[j].Filename
	})
}

func assetContent(asset artifact.ExpectedAsset) (io.ReadCloser, error) {
	if len(asset.Content) > 0 {
		return io.NopCloser(bytes.NewReader(asset.Content)), nil
	}
	return os.Open(asset.Path)
}

func identity(asset artifact.ExpectedAsset) AssetIdentity {
	return AssetIdentity{Filename: asset.Filename, SHA256: asset.SHA256}
}