You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: saga.go
    path of file: ./repos/ergo/gen
    
    the code of the file until where you have to start completion: package gen

import (
	"fmt"
	"math"
	"sync"
	"time"

	"github.com/ergo-services/ergo/etf"
	"github.com/ergo-services/ergo/lib"
)

// SagaBehavior interface
type SagaBehavior interface {
	ServerBehavior

	//
	// Mandatory callbacks
	//

	// InitSaga
	InitSaga(process *SagaProcess, args ...etf.Term) (SagaOptions, error)

	// HandleTxNew invokes on a new TX receiving by this saga.
	HandleTxNew(process *SagaProcess, id SagaTransactionID, value interface{}) SagaStatus

	// HandleTxResult invoked on a receiving result from the next saga
	HandleTxResult(process *SagaProcess, id SagaTransactionID, from SagaNextID, result interface{}) SagaStatus

	// HandleTxCancel invoked on a request of transaction cancelation.
	HandleTxCancel(process *SagaProcess, id SagaTransactionID, reason string) SagaStatus

	//
	// Optional callbacks
	//

	// HandleTxDone invoked when the transaction is done on a saga where it was created.
	// It returns the final result and SagaStatus. The commit message will deliver the final
	// result to all participants of this transaction (if it has enabled the TwoPhaseCommit option).
	// Otherwise the final result will be ignored.
	HandleTxDone(process *SagaProcess, id SagaTransactionID, result interface{}) (interface{}, SagaStatus)

	// HandleTxInterim invoked if received interim result from the next hop
	HandleTxInterim(process *SagaProcess, id SagaTransactionID, from SagaNextID, interim interface{}) SagaStatus

	// HandleTxCommit invoked if TwoPhaseCommit option is enabled for the given TX.
	// All sagas involved in this TX receive a commit message with final value and invoke this callback.
	// The final result has a value returned by HandleTxDone on a Saga created this TX.
	HandleTxCommit(process *SagaProcess, id SagaTransactionID, final interface{}) SagaStatus

	//
	// Callbacks to handle result/interim from the worker(s)
	//

	// HandleJobResult
	HandleJobResult(process *SagaProcess, id SagaTransactionID, from SagaJobID, result interface{}) SagaStatus
	// HandleJobInterim
	HandleJobInterim(process *SagaProcess, id SagaTransactionID, from SagaJobID, interim interface{}) SagaStatus
	// HandleJobFailed
	HandleJobFailed(process *SagaProcess, id SagaTransactionID, from SagaJobID, reason string) SagaStatus

	//
	// Server's callbacks
	//

	// HandleStageCall this callback is invoked on ServerProcess.Call. This method is optional
	// for the implementation
	HandleSagaCall(process *SagaProcess, from ServerFrom, message etf.Term) (etf.Term, ServerStatus)
	// HandleStageCast this callback is invoked on ServerProcess.Cast. This method is optional
	// for the implementation
	HandleSagaCast(process *SagaProcess, message etf.Term) ServerStatus
	// HandleStageInfo this callback is invoked on Process.Send. This method is optional
	// for the implementation
	HandleSagaInfo(process *SagaProcess, message etf.Term) ServerStatus
	// HandleSagaDirect this callback is invoked on Process.Direct. This method is optional
	// for the implementation
	HandleSagaDirect(process *SagaProcess, ref etf.Ref, message interface{}) (interface{}, DirectStatus)
}

const (
	defaultHopLimit = math.MaxUint16
	defaultLifespan = 60
)

// SagaStatus
type SagaStatus error

var (
	SagaStatusOK   SagaStatus // nil
	SagaStatusStop SagaStatus = fmt.Errorf("stop")

	// internal

	ErrSagaTxEndOfLifespan   = fmt.Errorf("End of TX lifespan")
	ErrSagaTxNextTimeout     = fmt.Errorf("Next saga timeout")
	ErrSagaUnknown           = fmt.Errorf("Unknown saga")
	ErrSagaJobUnknown        = fmt.Errorf("Unknown job")
	ErrSagaTxUnknown         = fmt.Errorf("Unknown TX")
	ErrSagaTxCanceled        = fmt.Errorf("Tx is canceled")
	ErrSagaTxInProgress      = fmt.Errorf("Tx is still in progress")
	ErrSagaResultAlreadySent = fmt.Errorf("Result is already sent")
	ErrSagaNotAllowed        = fmt.Errorf("Operation is not allowed")
)

// Saga
type Saga struct {
	Server
}

// SagaTransactionOptions
type SagaTransactionOptions struct {
	// HopLimit defines a number of hop within the transaction. Default limit
	// is 0 (no limit).
	HopLimit uint
	// Lifespan defines a lifespan for the transaction in seconds. Default is 60.
	Lifespan uint

	// TwoPhaseCommit enables 2PC for the transaction. This option makes all
	// Sagas involved in this transaction invoke HandleCommit callback on them and
	// invoke HandleCommitJob callback on Worker processes once the transaction is finished.
	TwoPhaseCommit bool
}

// SagaOptions
type SagaOptions struct {
	// MaxTransactions defines the limit for the number of active transactions. Default: 0 (unlimited)
	MaxTransactions uint
	// Worker
	Worker SagaWorkerBehavior
}

// SagaProcess
type SagaProcess struct {
	ServerProcess
	options  SagaOptions
	behavior SagaBehavior

	// running transactions
	txs      map[SagaTransactionID]*SagaTransaction
	mutexTXS sync.Mutex

	// next sagas where txs were sent
	next      map[SagaNextID]*SagaTransaction
	mutexNext sync.Mutex

	// running jobs
	jobs      map[etf.Pid]*SagaJob
	mutexJobs sync.Mutex
}

// SagaTransactionID
type SagaTransactionID etf.Ref

// String
func (
    
func (sp *SagaProcess) handleSagaRequest(m messageSaga) error {

	switch m.Request {
	case etf.Atom("$saga_next"):
		nextMessage := messageSagaNext{}

		if err := etf.TermIntoStruct(m.Command, &nextMessage); err != nil {
			return lib.ErrUnsupportedRequest
		}

		// Check if exceed the number of transaction on this saga
		if sp.options.MaxTransactions > 0 && len(sp.txs)+1 > int(sp.options.MaxTransactions) {
			cancel := etf.Tuple{
				etf.Atom("$saga_cancel"),
				sp.Self(),
				etf.Tuple{
					nextMessage.TransactionID,
					nextMessage.Origin,
					"exceed_tx_limit",
				},
			}
			sp.Send(m.Pid, cancel)
			return nil
		}

		// Check for the loop
		transactionID := SagaTransactionID(nextMessage.TransactionID)
		sp.mutexTXS.Lock()
		tx, ok := sp.txs[transactionID]
		sp.mutexTXS.Unlock()
		if ok {
			// loop detected. send cancel message
			cancel := etf.Tuple{
				etf.Atom("$saga_cancel"),
				sp.Self(),
				etf.Tuple{
					nextMessage.TransactionID,
					nextMessage.Origin,
					"loop_detected",
				},
			}
			sp.Send(m.Pid, cancel)
			return nil
		}

		txOptions := SagaTransactionOptions{
			HopLimit: defaultHopLimit,
			Lifespan: defaultLifespan,
		}
		if value, ok := nextMessage.Options["HopLimit"]; ok {
			if hoplimit, ok := value.(int64); ok {
				txOptions.HopLimit = uint(hoplimit)
			}
		}
		if value, ok := nextMessage.Options["Lifespan"]; ok {
			if lifespan, ok := value.(int64); ok && lifespan > 0 {
				txOptions.Lifespan = uint(lifespan)
			}
		}
		if value, ok := nextMessage.Options["TwoPhaseCommit"]; ok {
			txOptions.TwoPhaseCommit, _ = value.(bool)
		}

		tx = &SagaTransaction{
			id:      transactionID,
			options: txOptions,
			origin:  SagaNextID(nextMessage.Origin),
			next:    make(map[SagaNextID]*SagaNext),
			jobs:    make(map[SagaJobID]etf.Pid),
			arrival: time.Now().Unix(),
			parents: append([]etf.Pid{m.Pid}, nextMessage.Parents...),
		}
		sp.mutexTXS.Lock()
		sp.txs[transactionID] = tx
		sp.mutexTXS.Unlock()

		// do not monitor itself (they are equal if its came from the StartTransaction call)
		if m.Pid != sp.Self() {
			tx.monitor = sp.MonitorProcess(m.Pid)
		}

		// tx lifespan timer
		cancelMessage := etf.Tuple{
			etf.Atom("$saga_cancel"),
			sp.Self(), // can't be trapped (ignored)
			etf.Tuple{
				nextMessage.TransactionID,
				nextMessage.Origin,
				"lifespan",
			},
		}
		timeout := time.Duration(txOptions.Lifespan) * time.Second
		tx.cancelTimer = sp.SendAfter(sp.Self(), cancelMessage, timeout)

		return sp.behavior.HandleTxNew(sp, transactionID, nextMessage.Value)

	case "$saga_cancel":
		cancel := messageSagaCancel{}
		if err := etf.TermIntoStruct(m.Command, &cancel); err != nil {
			return lib.ErrUnsupportedRequest
		}

		tx, exist := sp.txs[SagaTransactionID(cancel.TransactionID)]
		if !exist {
			// unknown tx, just ignore it
			return nil
		}

		// check where it came from.
		if tx.parents[0] == m.Pid {
			// came from parent saga or from itself via CancelTransaction
			// can't be ignored
			sp.cancelTX(m.Pid, cancel, tx)
			return sp.behavior.HandleTxCancel(sp, tx.id, cancel.Reason)
		}

		// this cancel came from one of the next sagas
		// or from itself (being in the middle of transaction graph)
		next_id := SagaNextID(cancel.Origin)
		tx.Lock()
		next, ok := tx.next[next_id]
		tx.Unlock()

		if ok && next.TrapCancel {
			// clean the next saga stuff
			next.cancelTimer()
			sp.DemonitorProcess(cancel.Origin)
			tx.Lock()
			delete(tx.next, next_id)
			tx.Unlock()
			sp.mutexNext.Lock()
			delete(sp.next, next_id)
			sp.mutexNext.Unlock()

			// came from the next saga and TrapCancel was enabled
			cm := MessageSagaCancel{
				TransactionID: tx.id,
				NextID:        next_id,
				Reason:        cancel.Reason,
			}
			sp.Send(sp.Self(), cm)
			return SagaStatusOK
		}

		sp.cancelTX(m.Pid, cancel, tx)
		return sp.behavior.HandleTxCancel(sp, tx.id, cancel.Reason)

	case etf.Atom("$saga_result"):
		result := messageSagaResult{}
		if err := etf.TermIntoStruct(m.Command, &result); err != nil {
			return lib.ErrUnsupportedRequest
		}

		transactionID := SagaTransactionID(result.TransactionID)
		sp.mutexTXS.Lock()
		tx, ok := sp.txs[transactionID]
		sp.mutexTXS.Unlock()
		if !ok {
			// ignore unknown TX
			return nil
		}

		next_id := SagaNextID(result.Origin)
		empty_next_id := SagaNextID{}
		// next id is empty if we got result on a saga created this TX
		if next_id != empty_next_id {
			sp.mutexNext.Lock()
			_, ok := sp.next[next_id]
			sp.mutexNext.Unlock()
			if !ok {
				// ignore unknown result
				return nil
			}
			sp.mutexNext.Lock()
			delete(sp.next, next_id)
			sp.mutexNext.Unlock()

			tx.Lock()
			next := tx.next[next_id]
			if tx.options.TwoPhaseCommit == false {
				next.cancelTimer()
				sp.DemonitorProcess(result.Origin)
				delete(tx.next, next_id)
			} else {
				next.done = true
			}
			tx.Unlock()

			return sp.behavior.HandleTxResult(sp, tx.id, next_id, result.Result)
		}

		final, status := sp.behavior.HandleTxDone(sp, tx.id, result.Result)
		if status == SagaStatusOK {
			sp.commitTX(tx, final)
		}

		return status

	case etf.Atom("$saga_interim"):
		interim := messageSagaResult{}
		if err := etf.TermIntoStruct(m.Command, &interim); err != nil {
			return lib.ErrUnsupportedRequest
		}
		next_id := SagaNextID(interim.Origin)
		sp.mutexNext.Lock()
		tx, ok := sp.next[next_id]
		sp.mutexNext.Unlock()
		if !ok {
			// ignore unknown interim result and send cancel message to the sender
			message := etf.Tuple{
				etf.Atom("$saga_cancel"),
				sp.Self(),
				etf.Tuple{
					interim.TransactionID,
					interim.Origin,
					"unknown or canceled tx",
				},
			}
			sp.Send(m.Pid, message)
			return nil
		}
		return sp.behavior.HandleTxInterim(sp, tx.id, next_id, interim.Result)

	case etf.Atom("$saga_commit"):
		// propagate Commit signal if 2PC is enabled
		commit := messageSagaCommit{}
		if err := etf.TermIntoStruct(m.Command, &commit); err != nil {
			return lib.ErrUnsupportedRequest
		}
		transactionID := SagaTransactionID(commit.TransactionID)
		sp.mutexTXS.Lock()
		tx, ok := sp.txs[transactionID]
		sp.mutexTXS.Unlock()
		if !ok {
			// ignore unknown TX
			return nil
		}
		// clean up and send commit message before we invoke callback
		sp.commitTX(tx, commit.Final)
		// make sure if 2PC was enabled on this TX
		if tx.options.TwoPhaseCommit {
			return sp.behavior.HandleTxCommit(sp, tx.id, commit.Final)
		}
		return SagaStatusOK
	}
	return lib.ErrUnsupportedRequest
}