package mysql

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	"git.yuntuo.hk/k2safe/orbit/internal/jsonwire"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/platform/auth"
	"git.yuntuo.hk/k2safe/orbit/pulse/schedule"
	driver "github.com/go-sql-driver/mysql"
)

// ScheduleStore 管理计划控制与扫描；共享任务存储的事务认证，不在事务外推进游标。
type ScheduleStore struct {
	*TaskStore
	node core.ID
}

func NewScheduleStore(db *sql.DB, node core.ID) (*ScheduleStore, error) {
	if node.Validate() != nil {
		return nil, errors.New("scheduler node id required")
	}
	return &ScheduleStore{TaskStore: NewTaskStore(db), node: node}, nil
}

var _ schedule.Store = (*ScheduleStore)(nil)

const scheduleColumns = `id,namespace_id,queue_id,name,revision,current_version,processing_version,paused,next_fire_at,check_at,created_at,updated_at`

func scanSchedule(row scanner) (schedule.Summary, error) {
	var v schedule.Summary
	err := row.Scan(&v.ID, &v.NamespaceID, &v.QueueID, &v.Name, &v.Revision, &v.CurrentVersion, &v.ProcessingVersion, &v.Paused, &v.NextFireAt, &v.CheckAt, &v.CreatedAt, &v.UpdatedAt)
	return v, err
}
func readSchedule(ctx context.Context, tx *sql.Tx, ns, id core.ID, lock bool) (schedule.Summary, error) {
	q := `SELECT ` + scheduleColumns + ` FROM pulse_schedules WHERE namespace_id=? AND id=? AND queue_id IS NOT NULL`
	if lock {
		q += ` FOR UPDATE`
	}
	v, err := scanSchedule(tx.QueryRowContext(ctx, q, ns, id))
	if errors.Is(err, sql.ErrNoRows) {
		err = fault.NotFound
	}
	return v, err
}
func readScheduleVersion(ctx context.Context, tx *sql.Tx, ns, id core.ID, version uint64) (schedule.VersionView, error) {
	var v schedule.VersionView
	var rule, task []byte
	err := tx.QueryRowContext(ctx, `SELECT version,effective_at,specification,task_template,misfire_policy,overlap_policy,catchup_limit,catchup_window_ms,grace_ms FROM pulse_schedule_versions WHERE namespace_id=? AND schedule_id=? AND version=?`, ns, id, version).Scan(&v.Version, &v.EffectiveAt, &rule, &task, &v.Definition.Policy.Misfire, &v.Definition.Policy.Overlap, &v.Definition.Policy.CatchupLimit, &v.Definition.Policy.CatchupWindowMS, &v.Definition.Policy.GraceMS)
	if errors.Is(err, sql.ErrNoRows) {
		return v, fault.NotFound
	}
	if err != nil {
		return v, err
	}
	if err = json.Unmarshal(rule, &v.Definition.Rule); err != nil {
		return v, err
	}
	err = json.Unmarshal(task, &v.Definition.Task)
	return v, err
}
func insertScheduleVersion(ctx context.Context, tx *sql.Tx, ns, id core.ID, v schedule.VersionView) error {
	identity, err := core.NewID()
	if err != nil {
		return err
	}
	rule, err := json.Marshal(v.Definition.Rule)
	if err != nil {
		return err
	}
	task, err := jsonwire.Marshal(v.Definition.Task)
	if err != nil {
		return err
	}
	p := v.Definition.Policy
	_, err = tx.ExecContext(ctx, `INSERT INTO pulse_schedule_versions(namespace_id,id,schedule_id,version,effective_at,timezone,specification,task_template,misfire_policy,overlap_policy,catchup_limit,catchup_window_ms,grace_ms) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)`, ns, identity, id, v.Version, v.EffectiveAt, v.Definition.Rule.Timezone, string(rule), string(task), p.Misfire, p.Overlap, p.CatchupLimit, p.CatchupWindowMS, p.GraceMS)
	return err
}
func scheduleView(ctx context.Context, tx *sql.Tx, ns, id core.ID) (schedule.View, error) {
	summary, err := readSchedule(ctx, tx, ns, id, false)
	if err != nil {
		return schedule.View{}, err
	}
	current, err := readScheduleVersion(ctx, tx, ns, id, summary.CurrentVersion)
	return schedule.View{Summary: summary, Current: current}, err
}
func authorizeSchedule(ctx context.Context, tx *sql.Tx, ns, id core.ID, p auth.Permissions) (schedule.Summary, error) {
	v, err := readSchedule(ctx, tx, ns, id, false)
	if err != nil {
		return v, err
	}
	if !p.AllowsQueue(v.QueueID) {
		return v, fault.NotFound
	}
	return v, nil
}
func (s *ScheduleStore) Create(ctx context.Context, t auth.Token, key string, in schedule.CreateInput) (schedule.Result, error) {
	var out schedule.Result
	err := s.transaction(ctx, t, "schedules:manage", func(tx *sql.Tx, p auth.Permissions) error {
		if !p.Allows("tasks:submit") || !p.AllowsQueue(in.Definition.Task.QueueID) {
			return fault.Forbidden
		}
		id, err := core.NewID()
		if err != nil {
			return err
		}
		identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "schedule.create", "", key, "schedule", id, in)
		if err != nil {
			return err
		}
		if cached != nil {
			if err := json.Unmarshal(cached, &out); err != nil {
				return err
			}
			out.Deduplicated = true
			return nil
		}
		var queue core.ID
		err = tx.QueryRowContext(ctx, `SELECT id FROM pulse_queues WHERE namespace_id=? AND id=? FOR SHARE`, t.NamespaceID, in.Definition.Task.QueueID).Scan(&queue)
		if errors.Is(err, sql.ErrNoRows) {
			return fault.NotFound
		}
		if err != nil {
			return err
		}
		now, err := dbNow(ctx, tx)
		if err != nil {
			return err
		}
		c, err := schedule.Compile(in.Definition.Rule)
		if err != nil {
			return err
		}
		next, err := c.AtOrAfter(now)
		if err != nil {
			return err
		}
		if next == nil {
			return fault.Invalid("schedule has no future occurrence")
		}
		_, err = tx.ExecContext(ctx, `INSERT INTO pulse_schedules(namespace_id,id,queue_id,name,current_version,processing_version,paused,next_fire_at,check_at,created_at,updated_at) VALUES (?,?,?,?,1,1,?,?,?,?,?)`, t.NamespaceID, id, queue, in.Name, in.Paused, next, next, now, now)
		var me *driver.MySQLError
		if errors.As(err, &me) && me.Number == 1062 {
			return fault.Conflict
		}
		if err != nil {
			return err
		}
		if err := insertScheduleVersion(ctx, tx, t.NamespaceID, id, schedule.VersionView{Version: 1, EffectiveAt: now, Definition: in.Definition}); err != nil {
			return err
		}
		out.Schedule, err = scheduleView(ctx, tx, t.NamespaceID, id)
		if err != nil {
			return err
		}
		if err := emitEvent(ctx, tx, t.NamespaceID, id, 1, "schedule", "schedule.created", map[string]any{"queue_id": queue, "version": "1", "paused": in.Paused}); err != nil {
			return err
		}
		if err := appendAudit(ctx, tx, t, identity, id, "schedule", "schedule.create", "create schedule", nil, out.Schedule.Summary); err != nil {
			return err
		}
		return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
	})
	return out, err
}
func (s *ScheduleStore) Get(ctx context.Context, t auth.Token, id core.ID) (schedule.View, error) {
	var out schedule.View
	err := s.transaction(ctx, t, "schedules:read", func(tx *sql.Tx, p auth.Permissions) error {
		if _, err := authorizeSchedule(ctx, tx, t.NamespaceID, id, p); err != nil {
			return err
		}
		var err error
		out, err = scheduleView(ctx, tx, t.NamespaceID, id)
		return err
	})
	return out, err
}
func (s *ScheduleStore) Publish(ctx context.Context, t auth.Token, id core.ID, key string, in schedule.PublishInput) (schedule.Result, error) {
	return s.mutate(ctx, t, id, key, "schedule.publish", in, in.ExpectedRevision, in.Reason, func(tx *sql.Tx, v schedule.Summary, now time.Time) error {
		if in.Definition.Task.QueueID != v.QueueID {
			return fault.Invalid("schedule queue cannot change")
		}
		last, err := readScheduleVersion(ctx, tx, t.NamespaceID, id, v.CurrentVersion)
		if err != nil {
			return err
		}
		if !now.After(last.EffectiveAt) {
			return fault.Unavailable
		}
		if err := insertScheduleVersion(ctx, tx, t.NamespaceID, id, schedule.VersionView{Version: v.CurrentVersion + 1, EffectiveAt: now, Definition: in.Definition}); err != nil {
			return err
		}
		cursor, next := v.ProcessingVersion, v.NextFireAt
		// nil 游标也可能表示刚进入最新版本、尚未初始化；只有没有扫描工作才表示结束。
		if (next == nil && cursor == v.CurrentVersion && v.CheckAt == nil) || (next != nil && !next.Before(now)) {
			c, err := schedule.Compile(in.Definition.Rule)
			if err != nil {
				return err
			}
			next, err = c.AtOrAfter(now)
			if err != nil {
				return err
			}
			cursor = v.CurrentVersion + 1
		}
		check := &now
		if next != nil && next.After(now) {
			check = next
		} else if next == nil && cursor == v.CurrentVersion+1 {
			check = nil
		}
		_, err = tx.ExecContext(ctx, `UPDATE pulse_schedules SET current_version=current_version+1,processing_version=?,next_fire_at=?,check_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, cursor, next, check, now, t.NamespaceID, id)
		return err
	})
}
func (s *ScheduleStore) Control(ctx context.Context, t auth.Token, id core.ID, key string, in schedule.ControlInput, paused bool) (schedule.Result, error) {
	operation := "schedule.resume"
	if paused {
		operation = "schedule.pause"
	}
	return s.mutate(ctx, t, id, key, operation, in, in.ExpectedRevision, in.Reason, func(tx *sql.Tx, v schedule.Summary, now time.Time) error {
		if v.Paused == paused {
			return fault.Conflict
		}
		_, err := tx.ExecContext(ctx, `UPDATE pulse_schedules SET paused=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, paused, now, t.NamespaceID, id)
		return err
	})
}
func (s *ScheduleStore) mutate(ctx context.Context, t auth.Token, id core.ID, key, operation string, input any, revision uint64, reason string, change func(*sql.Tx, schedule.Summary, time.Time) error) (schedule.Result, error) {
	var out schedule.Result
	err := s.transaction(ctx, t, "schedules:manage", func(tx *sql.Tx, p auth.Permissions) error {
		if operation == "schedule.publish" && !p.Allows("tasks:submit") {
			return fault.Forbidden
		}
		if _, err := authorizeSchedule(ctx, tx, t.NamespaceID, id, p); err != nil {
			return err
		}
		identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, operation, string(id), key, "schedule", id, input)
		if err != nil {
			return err
		}
		if cached != nil {
			if err := json.Unmarshal(cached, &out); err != nil {
				return err
			}
			out.Deduplicated = true
			return nil
		}
		v, err := readSchedule(ctx, tx, t.NamespaceID, id, true)
		if err != nil {
			return err
		}
		if v.Revision != revision {
			return fault.Conflict
		}
		now, err := dbNow(ctx, tx)
		if err != nil {
			return err
		}
		if err := change(tx, v, now); err != nil {
			return err
		}
		out.Schedule, err = scheduleView(ctx, tx, t.NamespaceID, id)
		if err != nil {
			return err
		}
		if err := appendAudit(ctx, tx, t, identity, id, "schedule", operation, reason, v, out.Schedule.Summary); err != nil {
			return err
		}
		if err := emitEvent(ctx, tx, t.NamespaceID, id, out.Schedule.Revision, "schedule", operation, map[string]any{"version": fmt.Sprint(out.Schedule.CurrentVersion), "paused": out.Schedule.Paused}); err != nil {
			return err
		}
		return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
	})
	return out, err
}
func (s *ScheduleStore) Version(ctx context.Context, t auth.Token, id core.ID, version uint64) (schedule.VersionView, error) {
	var out schedule.VersionView
	err := s.transaction(ctx, t, "schedules:read", func(tx *sql.Tx, p auth.Permissions) error {
		if _, err := authorizeSchedule(ctx, tx, t.NamespaceID, id, p); err != nil {
			return err
		}
		var err error
		out, err = readScheduleVersion(ctx, tx, t.NamespaceID, id, version)
		return err
	})
	return out, err
}