package datastore

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"time"
)

// StatAppendLogTarget returns the minimal metadata needed by the append-log
// tail-append path in a single narrow query. It mirrors Stat's error
// semantics: a missing dentry, a file dentry whose inode row is absent, or an
// inode whose status is not CONFIRMED or whose contents/semantic row is absent
// reports ErrNotFound; a directory reports no File. The semantic table is
// joined only for existence so heavy embedding columns are never projected.
func (s *Store) StatAppendLogTarget(ctx context.Context, path string) (out *NodeWithFile, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "stat_append_log_target", start, &err)

	row := s.db.QueryRowContext(ctx, `SELECT fn.node_id, fn.path, fn.parent_path, fn.name, fn.is_directory, fn.file_id, fn.inode_id, fn.created_at,
		i.inode_id, i.size_bytes, i.revision, c.inode_id, c.storage_type, c.content_layout, c.storage_ref, c.content_type, sem.inode_id
		FROM file_nodes fn
		LEFT JOIN inodes i ON `+s.scope.AndOn(`fn.file_id = i.inode_id AND i.status = 'CONFIRMED'`, "i")+`
		LEFT JOIN contents c ON `+s.scope.AndOn(`i.inode_id = c.inode_id`, "c")+`
		LEFT JOIN semantic sem ON `+s.scope.AndOn(`i.inode_id = sem.inode_id`, "sem")+`
		WHERE `+s.scope.AndAs("fn", `fn.path_hash = ? AND fn.path = ?`)+`
		LIMIT 1`, scopeWhereArgs(s.scope, 3, s.scope.Args(fileNodePathHash(path), path)...)...)
	out, err = scanNodeWithFileAppendLog(row)
	if err != nil {
		return nil, err
	}
	return out, nil
}

// scanNodeWithFileAppendLog scans the StatAppendLogTarget projection into a
// NodeWithFile whose File carries only the fields the append-log path reads.
// It intentionally skips mode/status and heavy semantic columns. Missing
// inode, contents, or semantic rows mirror Stat's ErrNotFound propagation.
func scanNodeWithFileAppendLog(s scanner) (*NodeWithFile, error) {
	var n FileNode
	var isDir int
	var nodeFileID, nodeInodeID sql.NullString
	var nodeCreatedAt time.Time

	var fFileID sql.NullString
	var fSizeBytes, fRevision sql.NullInt64
	var cInodeID, semInodeID sql.NullString
	var fStorageType, fContentLayout, fStorageRef, fContentType sql.NullString

	err := s.Scan(&n.NodeID, &n.Path, &n.ParentPath, &n.Name, &isDir, &nodeFileID, &nodeInodeID, &nodeCreatedAt,
		&fFileID, &fSizeBytes, &fRevision, &cInodeID, &fStorageType, &fContentLayout, &fStorageRef, &fContentType, &semInodeID)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}

	n.IsDirectory = isDir != 0
	n.FileID = nodeFileID.String
	n.InodeID = nodeInodeID.String
	n.CreatedAt = nodeCreatedAt.UTC()

	nf := &NodeWithFile{Node: n}
	if n.IsDirectory || n.FileID == "" {
		return nf, nil
	}
	if !fFileID.Valid {
		// Mirror Stat: GetFile propagates GetInode's sql.ErrNoRows as
		// ErrNotFound, keeping the caller's create-fallback semantics.
		return nil, ErrNotFound
	}
	if !cInodeID.Valid {
		// Mirror Stat: GetFile propagates GetContent's missing-row error as
		// ErrNotFound, so a corrupted split-table state fails closed.
		return nil, ErrNotFound
	}
	if !semInodeID.Valid {
		// Mirror Stat: GetFile propagates GetSemantic's missing-row error as
		// ErrNotFound before any storage mutation.
		return nil, ErrNotFound
	}
	nf.File = &File{
		FileID:        fFileID.String,
		StorageType:   StorageType(fStorageType.String),
		ContentLayout: ContentLayout(fContentLayout.String),
		StorageRef:    fStorageRef.String,
		ContentType:   fContentType.String,
		SizeBytes:     fSizeBytes.Int64,
		Revision:      fRevision.Int64,
	}
	return nf, nil
}

// AppendLogContentIfMatchTx advances a confirmed append-log file only when its
// complete logical storage identity still matches the locked StatForUpdateTx
// snapshot. The caller must have already made the physical S3 append at
// expectedSize. Reusing the locked snapshot avoids a second
// GetFileStorageMetaForUpdateTx read. The final transaction advances inodes,
// preserves the legacy files dual-write, and narrow-clears any persisted
// checksum/content_blob on the matching contents row so stat/HEAD never
// advertise a pre-append digest.
func (s *Store) AppendLogContentIfMatchTx(tx execer, live *NodeWithFile, expectedRevision, expectedSize int64, expectedRef string, appendSize int64) (int64, error) {
	if expectedRevision < 0 || expectedSize < 0 || appendSize < 0 {
		return 0, fmt.Errorf("append-log values must be non-negative")
	}
	const maxInt64 = int64(^uint64(0) >> 1)
	if expectedSize > maxInt64-appendSize {
		return 0, fmt.Errorf("append-log size overflow")
	}
	if live == nil || live.File == nil {
		// Mirror GetFileStorageMetaForUpdateTx: a missing confirmed inode or
		// contents row reported ErrNotFound.
		return 0, ErrNotFound
	}
	if live.File.StorageType == "" || live.File.ContentLayout == "" {
		// Mirror GetFileStorageMetaForUpdateTx's INNER JOIN on contents:
		// storage_type/content_layout are NOT NULL, so empty values mean the
		// contents row is absent, which reported ErrNotFound. A present row
		// with an empty storage_ref still falls through to the ref CAS below,
		// matching the old comparison.
		return 0, ErrNotFound
	}
	if live.File.StorageType != StorageS3 || live.File.ContentLayout != ContentLayoutAppendLog ||
		live.File.Revision != expectedRevision || live.File.SizeBytes != expectedSize || live.File.StorageRef != expectedRef {
		return 0, ErrRevisionConflict
	}
	fileID := live.File.FileID

	now := time.Now().UTC()
	newSize := expectedSize + appendSize
	newRevision := expectedRevision + 1
	if s.useLegacyFiles {
		res, err := tx.Exec(`UPDATE files SET size_bytes = ?, revision = ?, checksum_sha256 = NULL,
			confirmed_at = ? WHERE file_id = ? AND storage_type = ? AND content_layout = ?
			AND storage_ref = ? AND revision = ? AND size_bytes = ? AND status = 'CONFIRMED'`,
			newSize, newRevision, now, fileID, StorageS3, ContentLayoutAppendLog,
			expectedRef, expectedRevision, expectedSize)
		if err != nil {
			return 0, err
		}
		if n, err := res.RowsAffected(); err != nil || n == 0 {
			if err != nil {
				return 0, err
			}
			return 0, fmt.Errorf("append-log legacy CAS: %w", ErrRevisionConflict)
		}
	}
	res, err := tx.Exec(`UPDATE inodes SET size_bytes = ?, revision = ?, mtime = ?, confirmed_at = ?
		WHERE `+s.scope.And(`inode_id = ? AND revision = ? AND size_bytes = ? AND status = 'CONFIRMED'`),
		append([]any{newSize, newRevision, now, now}, s.scope.Args(fileID, expectedRevision, expectedSize)...)...)
	if err != nil {
		return 0, err
	}
	if n, err := res.RowsAffected(); err != nil || n == 0 {
		if err != nil {
			return 0, err
		}
		return 0, fmt.Errorf("append-log inode CAS: %w", ErrRevisionConflict)
	}
	_, err = tx.Exec(`UPDATE contents SET checksum_sha256 = NULL, content_blob = NULL
		WHERE `+s.scope.And(`inode_id = ? AND storage_type = ? AND content_layout = ? AND storage_ref = ?`),
		s.scope.Args(fileID, StorageS3, ContentLayoutAppendLog, expectedRef)...)
	if err != nil {
		return 0, err
	}
	return newRevision, nil
}

// ReplaceAppendLogRefIfMatchTx advances an append-log generation without
// changing its logical length. It is used after a divergent-prefix or
// TooManyParts rebase.
func (s *Store) ReplaceAppendLogRefIfMatchTx(tx execer, fileID string, expectedRevision, expectedSize int64, expectedRef, nextRef string) (int64, error) {
	return s.ReplaceAppendLogContentIfMatchTx(tx, fileID, expectedRevision, expectedSize, expectedRef, nextRef, expectedSize)
}

// ReplaceAppendLogContentIfMatchTx atomically switches an append-log object
// generation and its committed logical size.
func (s *Store) ReplaceAppendLogContentIfMatchTx(tx execer, fileID string, expectedRevision, expectedSize int64, expectedRef, nextRef string, nextSize int64) (int64, error) {
	if nextRef == "" {
		return 0, fmt.Errorf("replacement append-log ref is required")
	}
	if nextSize < 0 {
		return 0, fmt.Errorf("replacement append-log size must be non-negative")
	}
	meta, err := s.GetFileStorageMetaForUpdateTx(tx, fileID)
	if err != nil {
		return 0, err
	}
	if meta.StorageType != StorageS3 || meta.ContentLayout != ContentLayoutAppendLog || meta.Revision != expectedRevision || meta.SizeBytes != expectedSize || meta.StorageRef != expectedRef {
		return 0, ErrRevisionConflict
	}
	now := time.Now().UTC()
	nextRevision := expectedRevision + 1
	if s.useLegacyFiles {
		res, err := tx.Exec(`UPDATE files SET storage_ref = ?, storage_ref_hash = ?, size_bytes = ?, revision = ?, checksum_sha256 = NULL, confirmed_at = ?
			WHERE file_id = ? AND storage_type = ? AND content_layout = ? AND storage_ref = ? AND revision = ? AND size_bytes = ? AND status = 'CONFIRMED'`,
			nextRef, StorageRefHash(nextRef), nextSize, nextRevision, now, fileID, StorageS3, ContentLayoutAppendLog, expectedRef, expectedRevision, expectedSize)
		if err != nil {
			return 0, err
		}
		if n, err := res.RowsAffected(); err != nil || n == 0 {
			if err != nil {
				return 0, err
			}
			return 0, ErrRevisionConflict
		}
	}
	res, err := tx.Exec(`UPDATE inodes SET size_bytes = ?, revision = ?, mtime = ?, confirmed_at = ? WHERE `+s.scope.And(`inode_id = ? AND revision = ? AND size_bytes = ? AND status = 'CONFIRMED'`),
		append([]any{nextSize, nextRevision, now, now}, s.scope.Args(fileID, expectedRevision, expectedSize)...)...)
	if err != nil {
		return 0, err
	}
	if n, err := res.RowsAffected(); err != nil || n == 0 {
		if err != nil {
			return 0, err
		}
		return 0, ErrRevisionConflict
	}
	res, err = tx.Exec(`UPDATE contents SET storage_ref = ?, storage_ref_hash = ?, checksum_sha256 = NULL, content_blob = NULL
		WHERE `+s.scope.And(`inode_id = ? AND storage_type = ? AND content_layout = ? AND storage_ref = ?`),
		append([]any{nextRef, StorageRefHash(nextRef)}, s.scope.Args(fileID, StorageS3, ContentLayoutAppendLog, expectedRef)...)...)
	if err != nil {
		return 0, err
	}
	if n, err := res.RowsAffected(); err != nil || n == 0 {
		if err != nil {
			return 0, err
		}
		return 0, ErrRevisionConflict
	}
	return nextRevision, nil
}