package automation

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

// ErrAutomationBlobMetadataConflict means an immutable log sequence already has
// metadata that does not match the retried payload.
var ErrAutomationBlobMetadataConflict = errors.New("automation blob metadata conflict")

type BlobMetrics struct {
	LogChunks        int64
	Artifacts        int64
	ExpiredArtifacts int64
}

type AutomationBlobReference struct {
	Kind        string
	WorkspaceID string
	RunID       string
	ID          string
	ObjectKey   string
	SizeBytes   int64
	SHA256      string
}

// CreateLeasedJobLogChunk stores one log-chunk metadata row and event while the
// queue job is still leased by leaseOwner. The unique
// (job_run_id, stream, sequence) key makes exact retries idempotent.
func (s *Store) CreateLeasedJobLogChunk(ctx context.Context, jobID int64, leaseOwner string, chunk AutomationLogChunk, event AutomationEvent, now time.Time) (AutomationLogChunk, AutomationEvent, error) {
	if err := s.ensureDB(); err != nil {
		return AutomationLogChunk{}, AutomationEvent{}, err
	}
	var stored AutomationLogChunk
	var storedEvent AutomationEvent
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := verifyLeasedAutomationJobTx(tx, jobID, leaseOwner, now); err != nil {
			return err
		}
		if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&chunk).Error; err != nil {
			return fmt.Errorf("create automation log chunk metadata: %w", err)
		}
		if err := tx.Where("job_run_id = ? AND stream = ? AND sequence = ?", chunk.JobRunID, chunk.Stream, chunk.Sequence).First(&stored).Error; err != nil {
			return fmt.Errorf("get automation log chunk metadata: %w", err)
		}
		if !sameAutomationLogChunkBlob(stored, chunk) {
			return ErrAutomationBlobMetadataConflict
		}
		if err := createOrGetAutomationEventTx(tx, event, &storedEvent); err != nil {
			return err
		}
		return nil
	})
	if err != nil {
		return AutomationLogChunk{}, AutomationEvent{}, err
	}
	return stored, storedEvent, nil
}

// VerifyLeasedJob checks that a queue job is still leased by leaseOwner at now.
func (s *Store) VerifyLeasedJob(ctx context.Context, jobID int64, leaseOwner string, now time.Time) error {
	if err := s.ensureDB(); err != nil {
		return err
	}
	var job AutomationJob
	err := s.db.WithContext(ctx).Select("id").Where("id = ? AND status = ? AND lease_owner = ? AND lease_until >= ?", jobID, AutomationJobStatusLeased, strings.TrimSpace(leaseOwner), now).First(&job).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return gorm.ErrRecordNotFound
	}
	if err != nil {
		return fmt.Errorf("verify automation job lease: %w", err)
	}
	return nil
}

// CreateLeasedJobEvent appends one automation event while the queue job is still
// leased by leaseOwner.
func (s *Store) CreateLeasedJobEvent(ctx context.Context, jobID int64, leaseOwner string, event AutomationEvent, now time.Time) (AutomationEvent, error) {
	if err := s.ensureDB(); err != nil {
		return AutomationEvent{}, err
	}
	var storedEvent AutomationEvent
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := verifyLeasedAutomationJobTx(tx, jobID, leaseOwner, now); err != nil {
			return err
		}
		return createOrGetAutomationEventTx(tx, event, &storedEvent)
	})
	if err != nil {
		return AutomationEvent{}, err
	}
	return storedEvent, nil
}

// CreateLeasedJobArtifact stores one artifact metadata row and event while the
// queue job is still leased by leaseOwner.
func (s *Store) CreateLeasedJobArtifact(ctx context.Context, jobID int64, leaseOwner string, artifact AutomationArtifact, event AutomationEvent, now time.Time) (AutomationArtifact, AutomationEvent, error) {
	if err := s.ensureDB(); err != nil {
		return AutomationArtifact{}, AutomationEvent{}, err
	}
	var storedEvent AutomationEvent
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := verifyLeasedAutomationJobTx(tx, jobID, leaseOwner, now); err != nil {
			return err
		}
		if err := tx.Create(&artifact).Error; err != nil {
			return fmt.Errorf("create automation artifact metadata: %w", err)
		}
		if err := createOrGetAutomationEventTx(tx, event, &storedEvent); err != nil {
			return err
		}
		return nil
	})
	if err != nil {
		return AutomationArtifact{}, AutomationEvent{}, err
	}
	return artifact, storedEvent, nil
}

func (s *Store) ListLogChunks(ctx context.Context, workspaceID string, runID string, jobRunID string, stream string, page int, perPage int) ([]AutomationLogChunk, error) {
	if err := s.ensureDB(); err != nil {
		return nil, err
	}
	offset := (page - 1) * perPage
	q := s.db.WithContext(ctx).Where("workspace_id = ? AND run_id = ?", strings.TrimSpace(workspaceID), strings.TrimSpace(runID))
	if trimmed := strings.TrimSpace(jobRunID); trimmed != "" {
		q = q.Where("job_run_id = ?", trimmed)
	}
	if trimmed := strings.TrimSpace(stream); trimmed != "" {
		q = q.Where("stream = ?", trimmed)
	}
	var chunks []AutomationLogChunk
	if err := q.Order("job_run_id ASC").Order("stream ASC").Order("sequence ASC").Limit(perPage).Offset(offset).Find(&chunks).Error; err != nil {
		return nil, fmt.Errorf("list automation log chunks: %w", err)
	}
	return chunks, nil
}

func (s *Store) GetLogChunk(ctx context.Context, workspaceID string, runID string, jobRunID string, stream string, sequence int64) (AutomationLogChunk, error) {
	if err := s.ensureDB(); err != nil {
		return AutomationLogChunk{}, err
	}
	var chunk AutomationLogChunk
	err := s.db.WithContext(ctx).Where("workspace_id = ? AND run_id = ? AND job_run_id = ? AND stream = ? AND sequence = ?", strings.TrimSpace(workspaceID), strings.TrimSpace(runID), strings.TrimSpace(jobRunID), strings.TrimSpace(stream), sequence).First(&chunk).Error
	if err != nil {
		return AutomationLogChunk{}, err
	}
	return chunk, nil
}

func (s *Store) ListArtifacts(ctx context.Context, workspaceID string, runID string, jobRunID string, page int, perPage int) ([]AutomationArtifact, error) {
	if err := s.ensureDB(); err != nil {
		return nil, err
	}
	offset := (page - 1) * perPage
	q := s.db.WithContext(ctx).Where("workspace_id = ? AND run_id = ?", strings.TrimSpace(workspaceID), strings.TrimSpace(runID))
	if trimmed := strings.TrimSpace(jobRunID); trimmed != "" {
		q = q.Where("job_run_id = ?", trimmed)
	}
	var artifacts []AutomationArtifact
	if err := q.Order("created_at ASC").Order("id ASC").Limit(perPage).Offset(offset).Find(&artifacts).Error; err != nil {
		return nil, fmt.Errorf("list automation artifacts: %w", err)
	}
	return artifacts, nil
}

func (s *Store) GetArtifact(ctx context.Context, workspaceID string, runID string, artifactID string) (AutomationArtifact, error) {
	if err := s.ensureDB(); err != nil {
		return AutomationArtifact{}, err
	}
	var artifact AutomationArtifact
	err := s.db.WithContext(ctx).Where("workspace_id = ? AND run_id = ? AND id = ?", strings.TrimSpace(workspaceID), strings.TrimSpace(runID), strings.TrimSpace(artifactID)).First(&artifact).Error
	if err != nil {
		return AutomationArtifact{}, err
	}
	return artifact, nil
}

func (s *Store) BlobMetrics(ctx context.Context, workspaceID string, now time.Time) (BlobMetrics, error) {
	if err := s.ensureDB(); err != nil {
		return BlobMetrics{}, err
	}
	var metrics BlobMetrics
	logs := s.db.WithContext(ctx).Model(&AutomationLogChunk{})
	artifacts := s.db.WithContext(ctx).Model(&AutomationArtifact{})
	if workspaceID = strings.TrimSpace(workspaceID); workspaceID != "" {
		logs = logs.Where("workspace_id = ?", workspaceID)
		artifacts = artifacts.Where("workspace_id = ?", workspaceID)
	}
	if err := logs.Session(&gorm.Session{}).Count(&metrics.LogChunks).Error; err != nil {
		return BlobMetrics{}, fmt.Errorf("count automation log chunks: %w", err)
	}
	if err := artifacts.Session(&gorm.Session{}).Count(&metrics.Artifacts).Error; err != nil {
		return BlobMetrics{}, fmt.Errorf("count automation artifacts: %w", err)
	}
	if err := artifacts.Session(&gorm.Session{}).Where("expires_at IS NOT NULL AND expires_at <= ?", now).Count(&metrics.ExpiredArtifacts).Error; err != nil {
		return BlobMetrics{}, fmt.Errorf("count expired automation artifacts: %w", err)
	}
	return metrics, nil
}

func (s *Store) ListExpiredArtifacts(ctx context.Context, now time.Time, limit int) ([]AutomationArtifact, error) {
	if err := s.ensureDB(); err != nil {
		return nil, err
	}
	if limit <= 0 {
		limit = 100
	}
	var artifacts []AutomationArtifact
	if err := s.db.WithContext(ctx).
		Where("expires_at IS NOT NULL AND expires_at <= ?", now).
		Order("expires_at ASC").
		Order("id ASC").
		Limit(limit).
		Find(&artifacts).Error; err != nil {
		return nil, fmt.Errorf("list expired automation artifacts: %w", err)
	}
	return artifacts, nil
}

func (s *Store) DeleteArtifactMetadata(ctx context.Context, workspaceID string, runID string, artifactID string) (bool, error) {
	if err := s.ensureDB(); err != nil {
		return false, err
	}
	result := s.db.WithContext(ctx).
		Where("workspace_id = ? AND run_id = ? AND id = ?", strings.TrimSpace(workspaceID), strings.TrimSpace(runID), strings.TrimSpace(artifactID)).
		Delete(&AutomationArtifact{})
	if result.Error != nil {
		return false, fmt.Errorf("delete automation artifact metadata: %w", result.Error)
	}
	return result.RowsAffected > 0, nil
}

func (s *Store) ExistingAutomationBlobObjectKeys(ctx context.Context, objectKeys []string) (map[string]struct{}, error) {
	if err := s.ensureDB(); err != nil {
		return nil, err
	}
	seen := make(map[string]struct{}, len(objectKeys))
	keys := make([]string, 0, len(objectKeys))
	for _, key := range objectKeys {
		key = strings.TrimSpace(key)
		if key == "" {
			continue
		}
		if _, ok := seen[key]; ok {
			continue
		}
		seen[key] = struct{}{}
		keys = append(keys, key)
	}
	existing := make(map[string]struct{}, len(keys))
	if len(keys) == 0 {
		return existing, nil
	}
	var logKeys []string
	if err := s.db.WithContext(ctx).Model(&AutomationLogChunk{}).Where("object_key IN ?", keys).Pluck("object_key", &logKeys).Error; err != nil {
		return nil, fmt.Errorf("check automation log object metadata: %w", err)
	}
	for _, key := range logKeys {
		existing[key] = struct{}{}
	}
	var artifactKeys []string
	if err := s.db.WithContext(ctx).Model(&AutomationArtifact{}).Where("object_key IN ?", keys).Pluck("object_key", &artifactKeys).Error; err != nil {
		return nil, fmt.Errorf("check automation artifact object metadata: %w", err)
	}
	for _, key := range artifactKeys {
		existing[key] = struct{}{}
	}
	return existing, nil
}

func (s *Store) ListAutomationBlobReferences(ctx context.Context, limit int) ([]AutomationBlobReference, error) {
	if err := s.ensureDB(); err != nil {
		return nil, err
	}
	if limit <= 0 {
		limit = 100
	}
	if limit > 1000 {
		limit = 1000
	}
	logLimit := limit / 2
	if logLimit == 0 {
		logLimit = 1
	}
	artifactLimit := limit - logLimit
	if artifactLimit == 0 {
		artifactLimit = 1
	}
	refs := make([]AutomationBlobReference, 0, limit)
	var chunks []AutomationLogChunk
	if err := s.db.WithContext(ctx).Order("created_at ASC").Order("id ASC").Limit(logLimit).Find(&chunks).Error; err != nil {
		return nil, fmt.Errorf("list automation log blob references: %w", err)
	}
	for _, chunk := range chunks {
		refs = append(refs, AutomationBlobReference{Kind: "log", WorkspaceID: chunk.WorkspaceID, RunID: chunk.RunID, ID: chunk.ID, ObjectKey: chunk.ObjectKey, SizeBytes: chunk.SizeBytes, SHA256: chunk.SHA256})
	}
	var artifacts []AutomationArtifact
	if err := s.db.WithContext(ctx).Order("created_at ASC").Order("id ASC").Limit(artifactLimit).Find(&artifacts).Error; err != nil {
		return nil, fmt.Errorf("list automation artifact blob references: %w", err)
	}
	for _, artifact := range artifacts {
		refs = append(refs, AutomationBlobReference{Kind: "artifact", WorkspaceID: artifact.WorkspaceID, RunID: artifact.RunID, ID: artifact.ID, ObjectKey: artifact.ObjectKey, SizeBytes: artifact.SizeBytes, SHA256: artifact.SHA256})
	}
	if len(refs) > limit {
		refs = refs[:limit]
	}
	return refs, nil
}

func verifyLeasedAutomationJobTx(tx *gorm.DB, jobID int64, leaseOwner string, now time.Time) error {
	result := tx.Model(&AutomationJob{}).Where("id = ? AND status = ? AND lease_owner = ? AND lease_until >= ?", jobID, AutomationJobStatusLeased, strings.TrimSpace(leaseOwner), now).Update("updated_at", now)
	if result.Error != nil {
		return fmt.Errorf("verify automation job lease: %w", result.Error)
	}
	if result.RowsAffected > 0 {
		return nil
	}

	// MySQL/TiDB report changed rows, not matched rows, for UPDATE. A runner
	// can claim and immediately accept a job within the same timestamp second,
	// making the updated_at touch a no-op even though the lease is valid.
	var job AutomationJob
	if err := tx.Select("id").
		Where("id = ? AND status = ? AND lease_owner = ? AND lease_until >= ?", jobID, AutomationJobStatusLeased, strings.TrimSpace(leaseOwner), now).
		First(&job).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return gorm.ErrRecordNotFound
		}
		return fmt.Errorf("verify automation job lease: %w", err)
	}
	return nil
}

func createOrGetAutomationEventTx(tx *gorm.DB, event AutomationEvent, storedEvent *AutomationEvent) error {
	if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&event).Error; err != nil {
		return fmt.Errorf("create automation blob event: %w", err)
	}
	if err := tx.Where("id = ?", event.ID).First(storedEvent).Error; err != nil {
		return fmt.Errorf("get automation blob event: %w", err)
	}
	return nil
}

func sameAutomationLogChunkBlob(left AutomationLogChunk, right AutomationLogChunk) bool {
	return left.WorkspaceID == right.WorkspaceID &&
		left.RunID == right.RunID &&
		left.JobRunID == right.JobRunID &&
		left.Stream == right.Stream &&
		left.Sequence == right.Sequence &&
		left.ObjectKey == right.ObjectKey &&
		left.SizeBytes == right.SizeBytes &&
		left.SHA256 == right.SHA256 &&
		left.ContentType == right.ContentType
}