package db

import (
	"encoding/json"
	"time"

	"tangled.org/core/api/tangled"
	"tangled.org/core/eventstream"
	"tangled.org/core/notifier"
	"tangled.org/core/spindle/models"
	"tangled.org/core/tid"
)

func (d *DB) insertEvent(event eventstream.Event, n *notifier.Notifier) error {
	return eventstream.Insert(d, event, n)
}

func (d *DB) GetEvents(cursor int64, limit int) ([]eventstream.Event, error) {
	return eventstream.List(d, cursor, limit)
}

func (d *DB) EventHighWater() (int64, error) {
	return eventstream.HighWater(d)
}

func (d *DB) CreatePipelineEvent(rkey string, pipeline tangled.Pipeline, n *notifier.Notifier) error {
	eventJson, err := json.Marshal(pipeline)
	if err != nil {
		return err
	}
	event := eventstream.Event{
		Rkey:      rkey,
		Nsid:      tangled.PipelineNSID,
		EventJson: eventJson,
	}
	return d.insertEvent(event, n)
}

// statusEvent builds the eventstream row for a pipeline status. Created is left
// zero so insertEvent stamps the local clock.
func statusEvent(pipelineAtUri, workflow, status string, workflowError *string, exitCode *int64) (eventstream.Event, error) {
	s := tangled.PipelineStatus{
		CreatedAt: time.Now().Format(time.RFC3339),
		Error:     workflowError,
		ExitCode:  exitCode,
		Pipeline:  pipelineAtUri,
		Workflow:  workflow,
		Status:    status,
	}

	eventJson, err := json.Marshal(s)
	if err != nil {
		return eventstream.Event{}, err
	}

	return eventstream.Event{
		Rkey:      tid.TID(),
		Nsid:      tangled.PipelineStatusNSID,
		EventJson: eventJson,
	}, nil
}

func (d *DB) createStatusEvent(
	workflowId models.WorkflowId,
	statusKind models.StatusKind,
	workflowError *string,
	exitCode *int64,
	n *notifier.Notifier,
) error {
	event, err := statusEvent(string(workflowId.PipelineId.AtUri()), workflowId.Name, string(statusKind), workflowError, exitCode)
	if err != nil {
		return err
	}
	return d.insertEvent(event, n)
}

// InsertRelayedStatus re-authors a status row relayed from an executor into the
// mill's own eventstream. The mill assigns its own Created timestamp (via
// insertEvent passing Created=0) so relayed rows order correctly against the
// mill's cursor, and NotifyAll wakes the mill's /events subscribers exactly
// as a local status write would.
func (d *DB) InsertRelayedStatus(
	pipelineAtUri string,
	workflow string,
	status string,
	workflowError *string,
	exitCode *int64,
	n *notifier.Notifier,
) error {
	event, err := statusEvent(pipelineAtUri, workflow, status, workflowError, exitCode)
	if err != nil {
		return err
	}
	return d.insertEvent(event, n)
}

// CompleteOrphanMillLease makes the terminal event and lease deletion one
// durable transition; an event without its matching deletion would be replayed.
func (d *DB) CompleteOrphanMillLease(
	leaseID string,
	pipelineAtUri string,
	workflow string,
	status string,
	workflowError *string,
	exitCode *int64,
	n *notifier.Notifier,
) error {
	event, err := statusEvent(pipelineAtUri, workflow, status, workflowError, exitCode)
	if err != nil {
		return err
	}
	tx, err := d.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if err := eventstream.Insert(tx, event, nil); err != nil {
		return err
	}
	if _, err := tx.Exec(`delete from mill_leases where lease_id = ?`, leaseID); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return err
	}
	n.NotifyAll()
	return nil
}

func (d *DB) GetStatus(workflowId models.WorkflowId) (*tangled.PipelineStatus, error) {
	pipelineAtUri := workflowId.PipelineId.AtUri()

	var eventJson string
	err := d.QueryRow(
		`
		select
			event from events
		where
			nsid = ?
			and json_extract(event, '$.pipeline') = ?
			and json_extract(event, '$.workflow') = ?
		order by
			created desc
		limit
			1
		`,
		tangled.PipelineStatusNSID,
		string(pipelineAtUri),
		workflowId.Name,
	).Scan(&eventJson)

	if err != nil {
		return nil, err
	}

	var status tangled.PipelineStatus
	if err := json.Unmarshal([]byte(eventJson), &status); err != nil {
		return nil, err
	}

	return &status, nil
}

func (d *DB) StatusPending(workflowId models.WorkflowId, n *notifier.Notifier) error {
	return d.createStatusEvent(workflowId, models.StatusKindPending, nil, nil, n)
}

func (d *DB) StatusRunning(workflowId models.WorkflowId, n *notifier.Notifier) error {
	return d.createStatusEvent(workflowId, models.StatusKindRunning, nil, nil, n)
}

func (d *DB) StatusFailed(workflowId models.WorkflowId, workflowError string, exitCode int64, n *notifier.Notifier) error {
	return d.createStatusEvent(workflowId, models.StatusKindFailed, &workflowError, &exitCode, n)
}

func (d *DB) StatusCancelled(workflowId models.WorkflowId, workflowError string, exitCode int64, n *notifier.Notifier) error {
	return d.createStatusEvent(workflowId, models.StatusKindCancelled, &workflowError, &exitCode, n)
}

func (d *DB) StatusSuccess(workflowId models.WorkflowId, n *notifier.Notifier) error {
	return d.createStatusEvent(workflowId, models.StatusKindSuccess, nil, nil, n)
}

func (d *DB) StatusTimeout(workflowId models.WorkflowId, n *notifier.Notifier) error {
	return d.createStatusEvent(workflowId, models.StatusKindTimeout, nil, nil, n)
}
