package main

// 🤖 s3BlobStore — blobStore implementation #2 (#550 Phase 1): attachment
// bytes in an S3-compatible object store (Cloudflare R2 is the intended
// target — zero egress; see docs/tier-economics.md). Enabled per instance by
// LCHAT_BLOB_S3_BUCKET in its env; unset keeps today's on-disk layout, so
// self-hosters and the fleet flip independently, instance by instance.
//
// The store carries the same E2EE posture as disk: bytes are client-side
// ciphertext, keys/filenames/types never reach here, and object keys are the
// server-minted blob ids — the object store learns exactly what the local
// disk knew, nothing more.
//
// Shape decisions, in the order they matter:
//
//   - WRAPS the disk store as a fallback, never replaces it: blobs uploaded
//     before the flip live on disk, and open/remove consult S3 first, then
//     disk. New writes always land in S3, so the disk set only shrinks —
//     the migration sweep (#550 Phase 2) drains it and the fallback then
//     never fires. Marker-free by construction.
//   - Staged writes SPOOL TO DISK (the same temp-in-blobs-dir shape as the
//     disk writer) and upload on commit. S3 has no rename, but a completed
//     PUT is atomic — an object is fetchable only in full — so "invisible
//     until commit" holds with no staging keys to garbage-collect. The spool
//     also gives the PUT a known ContentLength (R2 requires one).
//   - Reads are CHUNKED RANGED GETs behind an io.ReadSeekCloser:
//     http.ServeContent seeks (size probe, Range requests — video scrubs)
//     and then reads in 32KB slices; fetching per slice would be an S3
//     request per 32KB. The reader fetches aligned 512KB windows instead —
//     ~2 requests/MB served, and a scrub costs one window, not a full
//     download. (#836's lesson: request count is the S3 bill, watch it.)
//
// Config (instance env):
//
//	LCHAT_BLOB_S3_BUCKET    the switch — set = S3 store, unset = disk
//	LCHAT_BLOB_S3_PREFIX    key prefix; default = the DB's basename (the
//	                        host), so a shared fleet bucket needs no
//	                        per-instance wiring
//	LCHAT_BLOB_S3_REGION    optional ("auto" for R2)
//	LCHAT_BLOB_S3_ENDPOINT  optional custom endpoint (R2's
//	                        https://<account>.r2.cloudflarestorage.com);
//	                        implies path-style addressing
//
// Credentials ride the standard AWS chain (AWS_ACCESS_KEY_ID /
// AWS_SECRET_ACCESS_KEY in the instance env file — R2 issues S3-compatible
// pairs).

import (
	"context"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"os"
	"path"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/smithy-go"
)

// s3ReadChunk is the ranged-GET window. 512KB ≈ 2 requests per MB served;
// small enough that a video scrub's first paint isn't waiting on a huge
// window, large enough that request count stays a rounding error on the bill.
const s3ReadChunk = 512 << 10

// s3BlobTimeout bounds each individual S3 call (not a whole serve — a member
// streaming a long video holds the reader open far longer than any one GET).
const s3BlobTimeout = 60 * time.Second

type s3BlobStore struct {
	client *awss3.Client
	bucket string
	prefix string
	// fallback serves blobs the remote lacks (pre-flip disk blobs for the
	// primary store). Nil for a mirror store — a miss there is just a miss.
	fallback blobStore
	// spoolDir stages in-flight uploads — the disk store's dir, so the
	// temp-name convention ("." never appears in a valid blob id) keeps
	// staged bytes unfetchable exactly as before.
	spoolDir string
}

// s3BlobConfig reads the env switch. Nil config = feature off (disk store).
type s3BlobConfig struct {
	Bucket   string
	Prefix   string
	Region   string
	Endpoint string
	// Static credentials for THIS store — primary and mirror are different
	// providers with different key pairs, and the AWS default chain can only
	// hold one. Empty = the default chain (the primary's usual shape).
	AccessKey string
	SecretKey string
}

// s3BlobEnv reads one store's env quartet (+ key pair) under a prefix:
// LCHAT_BLOB_S3_* for the primary, LCHAT_BLOB_S3_MIRROR_* for the mirror.
func s3BlobEnv(envPrefix string) *s3BlobConfig {
	bucket := strings.TrimSpace(os.Getenv(envPrefix + "BUCKET"))
	if bucket == "" {
		return nil
	}
	return &s3BlobConfig{
		Bucket:    bucket,
		Prefix:    strings.Trim(strings.TrimSpace(os.Getenv(envPrefix+"PREFIX")), "/"),
		Region:    strings.TrimSpace(os.Getenv(envPrefix + "REGION")),
		Endpoint:  strings.TrimSpace(os.Getenv(envPrefix + "ENDPOINT")),
		AccessKey: strings.TrimSpace(os.Getenv(envPrefix + "ACCESS_KEY")),
		SecretKey: strings.TrimSpace(os.Getenv(envPrefix + "SECRET_KEY")),
	}
}

func s3BlobConfigFromEnv() *s3BlobConfig       { return s3BlobEnv("LCHAT_BLOB_S3_") }
func s3BlobMirrorConfigFromEnv() *s3BlobConfig { return s3BlobEnv("LCHAT_BLOB_S3_MIRROR_") }

// newS3BlobStore builds one remote store. fallback may be nil (a mirror);
// spoolDir stages writes; defaultPrefix is the DB basename — used when the
// env names none.
func newS3BlobStore(cfg *s3BlobConfig, spoolDir string, fallback blobStore, defaultPrefix string) (*s3BlobStore, error) {
	var opts []func(*awsconfig.LoadOptions) error
	if cfg.Region != "" {
		opts = append(opts, awsconfig.WithRegion(cfg.Region))
	}
	if cfg.AccessKey != "" {
		opts = append(opts, awsconfig.WithCredentialsProvider(
			credentials.NewStaticCredentialsProvider(cfg.AccessKey, cfg.SecretKey, "")))
	}
	awsCfg, err := awsconfig.LoadDefaultConfig(context.Background(), opts...)
	if err != nil {
		return nil, fmt.Errorf("blob s3: load aws config: %w", err)
	}
	client := awss3.NewFromConfig(awsCfg, func(o *awss3.Options) {
		if cfg.Endpoint != "" {
			o.BaseEndpoint = &cfg.Endpoint
			// Custom endpoints (R2, test fakes) address by path, not vhost.
			o.UsePathStyle = true
		}
		// The SDK's default-on flexible checksums wrap PutObject bodies in
		// aws-chunked framing, which R2 rejects; when-required sends plain
		// bodies (ContentLength is always known — the spool measured it).
		o.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
		o.ResponseChecksumValidation = aws.ResponseChecksumValidationWhenRequired
	})
	prefix := cfg.Prefix
	if prefix == "" {
		prefix = defaultPrefix
	}
	return &s3BlobStore{client: client, bucket: cfg.Bucket, prefix: prefix,
		fallback: fallback, spoolDir: spoolDir}, nil
}

func (s *s3BlobStore) key(id string) string { return path.Join(s.prefix, id) }

// s3NotFound folds the three shapes a missing object answers with (NoSuchKey
// on GET, NotFound on HEAD, and bare 404s from S3-compatibles) into one test.
func s3NotFound(err error) bool {
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		switch apiErr.ErrorCode() {
		case "NoSuchKey", "NotFound", "404":
			return true
		}
	}
	return false
}

func (s *s3BlobStore) create() (blobWriter, error) {
	f, err := os.CreateTemp(s.spoolDir, "up.*.part")
	if err != nil {
		return nil, err
	}
	return &s3BlobWriter{f: f, store: s}, nil
}

func (s *s3BlobStore) open(id string) (io.ReadSeekCloser, int64, time.Time, error) {
	ctx, cancel := context.WithTimeout(context.Background(), s3BlobTimeout)
	defer cancel()
	key := s.key(id)
	head, err := s.client.HeadObject(ctx, &awss3.HeadObjectInput{Bucket: &s.bucket, Key: &key})
	if err != nil {
		if s3NotFound(err) {
			if s.fallback != nil {
				// Pre-flip blob (or genuinely absent — the fallback answers
				// with fs.ErrNotExist either way, which is the contract).
				return s.fallback.open(id)
			}
			return nil, 0, time.Time{}, fs.ErrNotExist
		}
		return nil, 0, time.Time{}, err
	}
	size := int64(0)
	if head.ContentLength != nil {
		size = *head.ContentLength
	}
	mod := time.Time{}
	if head.LastModified != nil {
		mod = *head.LastModified
	}
	return &s3Reader{store: s, key: key, size: size}, size, mod, nil
}

func (s *s3BlobStore) remove(id string) error {
	ctx, cancel := context.WithTimeout(context.Background(), s3BlobTimeout)
	defer cancel()
	key := s.key(id)
	if _, err := s.client.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: &s.bucket, Key: &key}); err != nil && !s3NotFound(err) {
		return err
	}
	// A pre-flip blob's bytes live on disk — GC must reap those too. The
	// fallback treats a missing file as success, so this is free for S3-era ids.
	if s.fallback != nil {
		return s.fallback.remove(id)
	}
	return nil
}

type s3BlobWriter struct {
	f     *os.File
	store *s3BlobStore
}

func (w *s3BlobWriter) Write(p []byte) (int, error) { return w.f.Write(p) }

func (w *s3BlobWriter) commit(id string) error {
	defer os.Remove(w.f.Name())
	size, err := w.f.Seek(0, io.SeekEnd)
	if err != nil {
		w.f.Close()
		return err
	}
	if _, err := w.f.Seek(0, io.SeekStart); err != nil {
		w.f.Close()
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), s3BlobTimeout)
	defer cancel()
	key := w.store.key(id)
	_, err = w.store.client.PutObject(ctx, &awss3.PutObjectInput{
		Bucket:        &w.store.bucket,
		Key:           &key,
		Body:          w.f,
		ContentLength: &size,
	})
	w.f.Close()
	return err
}

func (w *s3BlobWriter) abort() {
	w.f.Close()
	os.Remove(w.f.Name())
}

// s3Reader is a lazy io.ReadSeekCloser over one object: seeks are position
// math (http.ServeContent's size probe costs nothing), reads fetch aligned
// s3ReadChunk windows on demand.
type s3Reader struct {
	store *s3BlobStore
	key   string
	size  int64
	pos   int64
	// The current window: buf holds object bytes [bufStart, bufStart+len(buf)).
	buf      []byte
	bufStart int64
}

func (r *s3Reader) Read(p []byte) (int, error) {
	if r.pos >= r.size {
		return 0, io.EOF
	}
	// Miss (or first read): fetch the aligned window containing pos.
	if r.buf == nil || r.pos < r.bufStart || r.pos >= r.bufStart+int64(len(r.buf)) {
		start := (r.pos / s3ReadChunk) * s3ReadChunk
		end := start + s3ReadChunk - 1 // Range is inclusive
		if end >= r.size {
			end = r.size - 1
		}
		ctx, cancel := context.WithTimeout(context.Background(), s3BlobTimeout)
		rng := fmt.Sprintf("bytes=%d-%d", start, end)
		out, err := r.store.client.GetObject(ctx, &awss3.GetObjectInput{
			Bucket: &r.store.bucket, Key: &r.key, Range: &rng,
		})
		if err != nil {
			cancel()
			if s3NotFound(err) {
				// Deleted mid-serve (GC race) — surface as the fs error the
				// handler already maps to 404.
				return 0, fs.ErrNotExist
			}
			return 0, err
		}
		buf, err := io.ReadAll(out.Body)
		out.Body.Close()
		cancel()
		if err != nil {
			return 0, err
		}
		r.buf, r.bufStart = buf, start
	}
	n := copy(p, r.buf[r.pos-r.bufStart:])
	r.pos += int64(n)
	return n, nil
}

func (r *s3Reader) Seek(offset int64, whence int) (int64, error) {
	var next int64
	switch whence {
	case io.SeekStart:
		next = offset
	case io.SeekCurrent:
		next = r.pos + offset
	case io.SeekEnd:
		next = r.size + offset
	default:
		return 0, fmt.Errorf("blob s3: bad whence %d", whence)
	}
	if next < 0 {
		return 0, fmt.Errorf("blob s3: negative seek")
	}
	r.pos = next
	return next, nil
}

func (r *s3Reader) Close() error {
	r.buf = nil
	return nil
}