/*
 * Package state provides an interface for a beanstalkd job, state for
 * job, tube and a connected client. and job state machine with the
 * states and transitions as defined in the protocol.
 * https://github.com/beanstalkd/beanstalkd/blob/master/doc/protocol.txt
 *
 * Implementations include an in-memory implementation of the interfaces
 */
package state

import (
	"fmt"
	log "github.com/sirupsen/logrus"
	"golang.org/x/net/context"
	"strings"
)

// JobState refers to the state of a job in beanstalkd context

//go:generate stringer -type=JobState --output state_string.go

type JobState int

const (
	Initial JobState = iota
	Ready
	Reserved
	Buried
	Delayed
	Deleted
)

type JobID uint64
type TubeName string
type ClientID string

type Job interface {
	// ID is a unique identifier integer for this job (generated by the server)
	ID() JobID

	// Priority is an integer < 2**32. Jobs with smaller priority values will be
	// scheduled before jobs with larger priorities. The most urgent priority is 0;
	// the least urgent priority is 4,294,967,295.
	Priority() uint32

	// UpdatePriority to a new value. Return back the newly set value
	UpdatePriority(newPriority uint32) uint32

	// Delay is an integer number of seconds to wait before putting the job in
	// the ready queue. The job will be in the "delayed" state during this time.
	// Maximum delay is 2**32-1.
	Delay() int64

	// Update Delay to a new value. Return back the newly set value
	UpdateDelay(newDelay int64) int64

	// TTR/time to run -- is an integer number of seconds to allow a worker
	// to run this job. This time is counted from the moment a worker reserves
	// this job. If the worker does not delete, release, or bury the job within
	// <ttr> seconds, the job will time out and the server will release the job.
	// The minimum ttr is 1. If the client sends 0, the server will silently
	// increase the ttr to 1. Maximum ttr is 2**32-1.
	TTR() int

	// BodySize is an integer indicating the size of the job body, not including the
	// trailing "\r\n". This value must be less than max-job-size (default: 2**16)
	BodySize() int

	// Body is the job body -- a sequence of bytes of length BodySize
	Body() []byte

	// TubeName the name of the tube associated with this job
	TubeName() TubeName

	// CreatedAt - Indicates the time, when this job is created
	CreatedAt() int64

	// ReadyAt - Indicates the time when the job is ready
	ReadyAt() int64

	// Reset the ReadyAt time taking the current time in reference
	// Return back the new readyAt time
	UpdateReadyAt(nowSeconds int64) (int64, error)

	// Retrieve the current job state
	State() JobState

	// Update the job state
	UpdateState(newState JobState)

	// Return the time at which the reservation expires
	ExpiresAt() int64

	// Reset the reservation time taking the current in reference
	// Return back the new reservation time
	UpdateReservation(nowSeconds int64) (int64, error)

	// ReservedBy returns the name of the client which has
	// reserved this job. Empty string if this job is not reserved
	ReservedBy() ClientID

	// Update the reservedBy client
	UpdateReservedBy(clientID ClientID)

	// Returns the time this specific job was buried
	BuriedAt() int64

	// Reset the buriedAt value to zero
	ResetBuriedAt()

	// Update the buriedAt value to the current clock
	// Return back the new BuriedAt time
	UpdateBuriedAt(nowSeconds int64) int64

	ReserveCount() uint32
	IncReserveCount()
	TimeoutCount() uint32
	IncTimeoutCount()
	ReleaseCount() uint32
	IncReleaseCount()
	BuryCount() uint32
	IncBuryCount()
	KickCount() uint32
	IncKickCount()
}

type localJob struct {
	nowSecs func() (int64, error)

	id         uint64
	priority   uint32
	delay      int64
	ttr        int
	bodySize   int
	body       []byte
	tubeName   TubeName
	createdAt  int64
	readyAt    int64
	state      JobState
	expiresAt  int64
	reservedBy ClientID
	buriedAt   int64

	reserveCount uint32
	timeoutCount uint32
	releaseCount uint32
	buryCount    uint32
	kickCount    uint32
}

func (j *localJob) ID() JobID {
	return JobID(j.id)
}

func (j *localJob) Priority() uint32 {
	return j.priority
}

func (j *localJob) UpdatePriority(newPriority uint32) uint32 {
	j.priority = newPriority
	return j.priority
}

func (j *localJob) Delay() int64 {
	return j.delay
}

func (j *localJob) UpdateDelay(newDelay int64) int64 {
	j.delay = newDelay
	return j.delay
}

func (j *localJob) TTR() int {
	return j.ttr
}

func (j *localJob) BodySize() int {
	return j.bodySize
}

func (j *localJob) Body() []byte {
	return j.body
}

func (j *localJob) TubeName() TubeName {
	return j.tubeName
}

func (j *localJob) CreatedAt() int64 {
	return j.createdAt
}

func (j *localJob) ReadyAt() int64 {
	return j.readyAt
}

func (j *localJob) UpdateReadyAt(nowSeconds int64) (int64, error) {
	j.readyAt = nowSeconds + j.delay
	return j.readyAt, nil
}

func (j *localJob) State() JobState {
	return j.state
}

func (j *localJob) UpdateState(newState JobState) {
	if j.state == Buried {
		j.buriedAt = 0
	}

	j.state = newState
}

func (j *localJob) UpdateReservedBy(clientID ClientID) {
	j.reservedBy = clientID
}

func (j *localJob) ExpiresAt() int64 {
	return j.expiresAt
}

func (j *localJob) ReservedBy() ClientID {
	return j.reservedBy
}

func (j *localJob) UpdateReservation(nowSeconds int64) (int64, error) {
	j.expiresAt = nowSeconds + int64(j.ttr)
	return j.expiresAt, nil
}

func (j *localJob) ResetBuriedAt() {
	j.buriedAt = 0
}

func (j *localJob) UpdateBuriedAt(nowSeconds int64) int64 {
	j.buriedAt = nowSeconds
	return j.buriedAt
}

func (j *localJob) BuriedAt() int64 {
	return j.buriedAt
}

func (j *localJob) ReserveCount() uint32 {
	return j.reserveCount
}
func (j *localJob) IncReserveCount() {
	j.reserveCount++
}
func (j *localJob) TimeoutCount() uint32 {
	return j.timeoutCount
}
func (j *localJob) IncTimeoutCount() {
	j.timeoutCount++
}
func (j *localJob) ReleaseCount() uint32 {
	return j.releaseCount
}
func (j *localJob) IncReleaseCount() {
	j.releaseCount++
}
func (j *localJob) BuryCount() uint32 {
	return j.buryCount
}
func (j *localJob) IncBuryCount() {
	j.buryCount++
}
func (j *localJob) KickCount() uint32 {
	return j.kickCount
}
func (j *localJob) IncKickCount() {
	j.kickCount++
}

func GetStatistics(nowSecs int64, j Job) map[string]interface{} {
	jobState := fmt.Sprintf("%s", j.State().String())
	age := nowSecs - j.CreatedAt()
	timeLeft := int64(0)
	if j.State() == Delayed {
		timeLeft = j.ReadyAt() - nowSecs
		if timeLeft < 0 {
			log.Errorf("GetStatistics: Delayed Job timeLeft=%v < 0. nowSecs = %v, ReadyAt = %v",
				timeLeft, nowSecs, j.ReadyAt())
			timeLeft = 0
		}
	}
	if j.State() == Reserved {
		timeLeft = j.ExpiresAt() - nowSecs
		if timeLeft < 0 {
			log.Errorf("GetStatistics: Reserved Job timeLeft=%v < 0. nowSecs = %v, ExpiresAt = %v",
				timeLeft, nowSecs, j.ExpiresAt())
			timeLeft = 0
		}
	}
	return map[string]interface{}{
		"id":        j.ID(),
		"tube":      j.TubeName(),
		"state":     strings.ToLower(jobState),
		"pri":       j.Priority(),
		"age":       age,
		"delay":     j.Delay(),
		"ttr":       j.TTR(),
		"time-left": timeLeft,
		"file":      0,
		"reserves":  j.ReserveCount(),
		"timeouts":  j.TimeoutCount(),
		"releases":  j.ReleaseCount(),
		"buries":    j.BuryCount(),
		"kicks":     j.KickCount(),
	}
}

// JSM provides methods for the beanstalkd job state machine.
// put with delay               release with delay
//  ----------------> [DELAYED] <------------.
//                        |                   |  touch (extend ttr)
//                        | timeout/ttr       | .----.
//                        |                   | |    |
//   put                  v     reserve       | |    v   delete
//  -----------------> [READY] ------------> [RESERVED] --------> *poof*
//                       ^  ^                | | |
//                       |  ^\  release      | | |
//                       |   \ `-------------' | |
//                       |    \                | |
//                       |     \  timeout/ttr  , |
//                       |      `--------------  |
//                       |                       |
//                       | kick                  |
//                       |                       |
//                       |       bury            |
//                    [BURIED] <-----------------'
//                       |
//                       |  delete
//                        `--------> *poof*
type JSM interface {
	// Put makes a new job. It initializes the job with
	// a unique identifier, sets state to Ready or Delayed based
	// on the delay parameter.
	Put(nowSeconds int64,
		priority uint32,
		delay int64,
		ttr int,
		bodySize int,
		body []byte,
		tubeName TubeName) (JobID, error)

	// Delete, removes a job specified by the id by a specific client
	Delete(jobID JobID, clientID ClientID) error

	// PeekDelayedJob returns the job in the Delay state in order of
	// priority for this tube. A job with and earlier (lower) delay
	// takes higher precedence.
	PeekDelayedJob(tubeName TubeName) (Job, error)

	// PeekReadyJob returns the job in the Ready state in order of
	// priority for this tube. A job with a lower priority value
	// takes higher precedence.
	PeekReadyJob(tubeName TubeName) (Job, error)

	// PeekReadyJob returns the job in the Buried state in order of
	// priority for this tube. A job which was buried first takes higher precedence.
	PeekBuriedJob(tubeName TubeName) (Job, error)

	// GetJob returns the job by its id
	GetJob(id JobID) (Job, error)

	// Release transitions this reserved job to a Ready state
	Release(jobID JobID, clientID ClientID) error

	// ReleaseWith transitions this reserved job to a Ready or Delayed state
	// with a modified priority and delay
	ReleaseWith(nowSeconds int64, jobID JobID, clientID ClientID, pri uint32, delay int64) error

	// Extend a reserved job's reservation TTL by its TTR (time-to-run)
	Touch(nowSeconds int64, jobID JobID, clientID ClientID) error

	// Bury this job (from a reserved state)
	Bury(nowSeconds int64, jobID JobID, priority uint32, clientID ClientID) error

	// Kick this job from buried state to a ready state
	Kick(jobID JobID) error

	// Kick atmost n jobs from the specified tube to ready state
	KickN(name TubeName, n int) (int, error)

	// Retrieve the statistics of this job
	// The stats-job data is a YAML byte slice representing a single dictionary of string
	// keys to scalar values. It contains these keys:
	// - "id" is the job id
	// - "tube" is the name of the tube that contains this job
	// - "state" is "ready" or "delayed" or "reserved" or "buried"
	// - "pri" is the priority value set by the put, release, or bury commands.
	// - "age" is the time in seconds since the put command that created this job.
	// - "delay" is the integer number of seconds to wait before putting this job in
	//   the ready queue.
	// - "ttr" -- time to run -- is the integer number of seconds a worker is
	//   allowed to run this job.
	// - "time-left" is the number of seconds left until the server puts this job
	//   into the ready queue. This number is only meaningful if the job is
	//   reserved or delayed. If the job is reserved and this amount of time
	//   elapses before its state changes, it is considered to have timed out.
	// - "file" this will be 0.
	// - "reserves" is the number of times this job has been reserved.
	// - "timeouts" is the number of times this job has timed out during a
	//   reservation.
	// - "releases" is the number of times a client has released this job from a
	//   reservation.
	// - "buries" is the number of times this job has been buried.
	// - "kicks" is the number of times this job has been kicked.
	GetStatsJobAsYaml(nowSeconds int64, id JobID) ([]byte, error)

	// Retrieve tube statistics
	//
	// 	The stats-job data is a YAML byte slice representing a single dictionary of string
	//	keys to scalar values. It contains these keys:
	//  - "name" is the tube's name.
	// - "current-jobs-ready" is the number of jobs in the ready queue in this tube.
	// - "current-jobs-reserved" is the number of jobs reserved by all clients in this tube.
	// - "current-jobs-delayed" is the number of delayed jobs in this tube.
	// - "current-jobs-buried" is the number of buried jobs in this tube.
	// - "current-waiting" is the number of open connections that have issued a
	//   reserve command while watching this tube but not yet received a response.
	//
	// The following are not implemented but have placeholders for backward-compat:
	// - "current-jobs-urgent" always zero
	// - "total-jobs" is always zero
	// - "current-using" is always zero
	// - "current-watching" is the number of open connections that are currently
	//   watching this tube.
	// - "pause" is the number of seconds the tube has been paused for.
	// - "cmd-delete" is the cumulative number of delete commands for this tube
	// - "cmd-pause-tube" is the cumulative number of pause-tube commands for this tube.
	// - "pause-time-left" is the number of seconds until the tube is un-paused.
	GetStatsTubeAsYaml(nowSeconds int64, tubeName TubeName) ([]byte, error)

	// Retrieve overall statistics
	GetStatsAsYaml(nowSeconds int64) ([]byte, error)

	// Returns an interface that allows a caller to snapshot the current
	// state of the JSM. Callers of the interface should not be done across
	// go-routines.
	Snapshot() (JSMSnapshot, error)

	// AppendReservation Appends a new Reservation Request for a client, and the specified set of tubes
	// if the timeoutSecs is zero, then an infinite timeout is assumed.
	//
	// A pointer to a Reservation struct is returned which encapsulates a result if a reservation
	// was handled or not
	AppendReservation(clientId ClientID, reqID string, watchedTubes []TubeName, nowSecs, deadlineAt int64) (*Reservation, error)

	// Tick runs a step of the job state machine with the current time.
	//
	// This call returns all the allocated reservations in this tick call
	Tick(nowSecs int64) ([]*Reservation, error)

	// GetTubes returns the tubeNames of all current tubes
	GetTubes() ([]TubeName, error)

	// CheckClientState queries the job state machine whether the provided list of clientIds are waiting for reservations.
	//
	// The response returns the ClientIDs (i) which are waiting for reservations, (ii) those which are not waiting and (iii)
	// those which are missing or an error.
	CheckClientState(clientIDs []ClientID) ([]ClientID, []ClientID, []ClientID, error)

	// Close or stop this jobstatemachine
	Stop() error
}

// JSMSnapshot provides methods allowing a caller to read & restore
// jobs  out of the job state machine.
type JSMSnapshot interface {
	// SnapshotJobs returns a read-only job channel. This allows a caller
	// to iterate through the jobs sent on the channel, when all the jobs
	// in the state machine are returned, this method closes the channel,
	// signaling the end of this snapshot.
	//
	// SnapshotJobs is used to support log compaction. This call should
	// an be used to save a point-in-time snapshot of the FSM.
	//
	// SnapshotJobs should not be called to the JSM across go-routines,
	// this is the default behavior (unless an implementation forces to override this)
	// A caller is recommended to clone this job
	SnapshotJobs() (<-chan Job, error)

	// SnapshotClients returns a read-only job channel. This allows a caller
	// to iterate through the clientResvEntries sent on the channel, when
	// all the entries in the state machine are returned, this method closes
	// the channel, signaling the end of this snapshot.
	//
	// SnapshotClients is used to support log compaction. This call should
	// an be used to save a point-in-time snapshot of the FSM.
	//
	// SnapshotClients should not be called to the JSM across go-routines,
	// this is the default behavior (unless an implementation forces to override this)
	// A caller is recommended to clone this job
	SnapshotClients() (<-chan *ClientResvEntry, error)

	// RestoreJobs takes jobsCh, a write-only job channel, that allow a caller
	// to send jobs to be added to the job state machine (JSM).
	//
	// Once RestoreJobs and RestoreClients are complete. call FinalizeRestore
	// which replaces the current state of JSM with the jobs provided in the
	// channel.
	//
	// RestoreJobs takes an additional context which can be used to signal
	// a cancellation. In this case, the method discards the jobs that were
	// provided on the channel, after the cancel is called
	//
	// Note: It is the responsibility of the caller to close the channels
	// and drain the jobsCh
	RestoreJobs(ctx context.Context, jobsCh <-chan Job) error

	// RestoreClients takes entriesCh, a write-only job channel, that allow
	// a caller to send clientResvEntry structs to be added to the job state
	// machine (JSM).
	//
	// Once RestoreJobs and RestoreClients are complete. call FinalizeRestore
	// which replaces the current state of JSM with the jobs provided in the
	// channel.
	//
	// RestoreClients takes an additional context which can be used to signal
	// a cancellation. In this case, the method discards the clientResvEntries
	// that were provided on the channel, after the cancel is called
	//
	// Note: It is the responsibility of the caller to close the channels
	// and drain the jobsCh
	RestoreClients(ctx context.Context, nClients int, entriesCh <-chan *ClientResvEntry) error

	// Finalize Restore overwrites the state of the job-state-machine
	// with the current state of the snapshot
	//
	// Once RestoreJobs and RestoreClients are complete. call FinalizeRestore
	// which replaces the current state of JSM with the jobs provided in the
	// channel.
	FinalizeRestore()
}

// Refers to the Reservation Status of a Reservation

// go:generate stringer -type=ReservationStatus --output state_string.go
type ReservationStatus int

const (
	Unknown ReservationStatus = iota
	Queued
	DeadlineSoon
	Matched
	Timeout
	Error
)

type Reservation struct {
	RequestId string
	ClientId  ClientID
	JobId     JobID
	Status    ReservationStatus
	BodySize  int
	Body      []byte
	Error     error
}

func (r Reservation) String() string {
	return fmt.Sprintf("Reservation ReqID=%s ClientID=%v Status=%v JobID=%v Error=%v BodySize=%v",
		r.RequestId, r.ClientId, r.Status, r.JobId, r.Error, r.BodySize)
}
