package hostops

import (
	"context"
	"fmt"
	"io"

	"github.com/carlosframework/platform/internal/geoip"
	"github.com/carlosframework/platform/internal/release"
)

var _ release.PutStreamer = (*mutationBoundary)(nil)

// PutFrom is release.PutStreamer, written out so the boundary is never
// bypassed by a streaming write and a large object is never buffered through
// it. Only a create-only put of an exact GeoIP blob key or an exact release
// rootfs image key streams: it is classified by key (a GeoIP blob was hashed
// by its publisher while staging, a rootfs image by the ship, and a
// create-only put cannot replace an object), bounded by the size its kind
// allows, then handed to the raw store. Every other key does what
// release.PutObjectFrom does for a store with no PutFrom: a bounded buffer and
// the ordinary checked Put.
func (s *mutationBoundary) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
	limit := streamLimit(key, ifMatch)
	if limit == 0 {
		if size > release.PutFromFallbackMax {
			return "", fmt.Errorf("%s: %w", key, release.ErrTooLargeForPut)
		}
		b, err := io.ReadAll(io.LimitReader(body, size+1))
		if err != nil {
			return "", err
		}
		if int64(len(b)) != size {
			return "", fmt.Errorf("%s: body is %d bytes, want %d", key, len(b), size)
		}
		return s.Put(ctx, key, b, ifMatch)
	}
	if size <= 0 || size > limit {
		err := fmt.Errorf("%w: %s: size %d out of range", ErrUnclassified, key, size)
		s.report(MutationTarget{Surface: "bucket", Method: "Put", Key: key}, err)
		return "", err
	}
	if err := s.decide(ctx, MutationTarget{Surface: "bucket", Method: "Put", Key: key}); err != nil {
		return "", err
	}
	MarkAppMutationIssued(ctx)
	return release.PutObjectFrom(ctx, s.raw, key, body, size, "")
}

// streamLimit is the largest body a create-only put of key may stream past
// the boundary, or 0 when key does not stream.
func streamLimit(key, ifMatch string) int64 {
	switch {
	case ifMatch != "":
		return 0
	case IsGeoipBlobKey(key):
		return geoip.MaxBlobBytes
	case release.IsRootfsArtifactKey(key):
		return release.MaxRootfsBytes
	}
	return 0
}

// PutFrom preserves the streaming seam beneath the admission boundary. Only
// immutable deployment GeoIP blobs and release rootfs images bypass
// record-body parsing (neither is ever a record body); app writes still use
// the conditional record checks and the outer mutation boundary.
func (s conditionalRecordStore) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
	if (IsGeoipBlobKey(key) || release.IsRootfsArtifactKey(key)) && ifMatch == "" {
		return release.PutObjectFrom(ctx, s.raw, key, body, size, ifMatch)
	}
	if size < 0 || size > release.PutFromFallbackMax {
		return "", release.ErrTooLargeForPut
	}
	b, err := io.ReadAll(io.LimitReader(body, size+1))
	if err != nil {
		return "", err
	}
	if int64(len(b)) != size {
		return "", ErrIdentity
	}
	return s.Put(ctx, key, b, ifMatch)
}