240 lines
8.8 KiB
Go
240 lines
8.8 KiB
Go
package writers
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// ExecutionReport describes an entire invocation, including operations never attempted.
|
|
// ExecutedStatements counts operations still applied, excluding confirmed rollbacks.
|
|
type ExecutionReport struct {
|
|
Dialect string `json:"dialect,omitempty"`
|
|
TotalStatements int `json:"total_statements"`
|
|
ExecutedStatements int `json:"executed_statements"`
|
|
FailedStatements int `json:"failed_statements"`
|
|
SkippedStatements int `json:"skipped_statements"`
|
|
RolledBackStatements int `json:"rolled_back_statements"`
|
|
Outcome string `json:"outcome"`
|
|
Operations []ExecutionOperation `json:"operations"`
|
|
Schemas []SchemaReport `json:"schemas"`
|
|
Errors []ExecutionError `json:"errors,omitempty"`
|
|
StartTime string `json:"start_time"`
|
|
EndTime string `json:"end_time"`
|
|
}
|
|
|
|
type SchemaReport struct {
|
|
Name string `json:"name"`
|
|
Tables []TableReport `json:"tables"`
|
|
}
|
|
|
|
type TableReport struct {
|
|
Name string `json:"name"`
|
|
Created bool `json:"created"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
type ExecutionError struct {
|
|
StatementNumber int `json:"statement_number"`
|
|
Statement string `json:"statement"`
|
|
Error string `json:"error"`
|
|
}
|
|
|
|
// ExecutionOperation is one generated statement or supplied script, in execution order.
|
|
// A failed raw script may have committed some of its own statements; see Outcome.
|
|
type ExecutionOperation struct {
|
|
StatementNumber int `json:"statement_number"`
|
|
Name string `json:"name,omitempty"`
|
|
Statement string `json:"statement"`
|
|
Status string `json:"status"` // applied, executed (raw script), failed, skipped, rolled_back, unknown
|
|
Error string `json:"error,omitempty"`
|
|
Skip bool `json:"-"`
|
|
StopOnFailure bool `json:"-"`
|
|
}
|
|
|
|
// ExecutionFailure is returned even when continuation was explicitly requested.
|
|
type ExecutionFailure struct{ Report *ExecutionReport }
|
|
|
|
func (e *ExecutionFailure) Error() string {
|
|
message := fmt.Sprintf("%s execution failed: %d failed, %d executed, %d skipped, %d rolled back",
|
|
e.Report.Dialect, e.Report.FailedStatements, e.Report.ExecutedStatements,
|
|
e.Report.SkippedStatements, e.Report.RolledBackStatements)
|
|
if len(e.Report.Errors) > 0 {
|
|
first := e.Report.Errors[0]
|
|
message += fmt.Sprintf("; statement %d: %s", first.StatementNumber, first.Error)
|
|
}
|
|
return message
|
|
}
|
|
|
|
// ExecutionBackend operates on one connection. Begin switches Exec to a transaction.
|
|
// Atomic execution stops at the first failure and rolls back the whole invocation.
|
|
type ExecutionBackend struct {
|
|
Exec func(context.Context, int, string) error
|
|
Begin func(context.Context) error
|
|
Commit func(context.Context) error
|
|
Rollback func(context.Context) error
|
|
// ScriptManaged means arbitrary scripts can commit/rollback internally.
|
|
ScriptManaged bool
|
|
}
|
|
|
|
func (o *WriterOptions) ShouldContinue() bool {
|
|
legacy, _ := o.Metadata["ignore_errors"].(bool)
|
|
return o.ContinueOnError || legacy
|
|
}
|
|
|
|
// RunExecution applies a plan, records every outcome, and publishes metadata and reports.
|
|
func RunExecution(ctx context.Context, opts *WriterOptions, report *ExecutionReport, operations []ExecutionOperation, backend ExecutionBackend) error {
|
|
report.TotalStatements = len(operations)
|
|
report.Operations = append([]ExecutionOperation(nil), operations...)
|
|
report.ExecutedStatements, report.FailedStatements, report.SkippedStatements, report.RolledBackStatements = 0, 0, 0, 0
|
|
report.Errors = nil
|
|
report.StartTime = time.Now().Format(time.RFC3339Nano)
|
|
report.Outcome = "not_started"
|
|
for i := range report.Operations {
|
|
report.Operations[i].StatementNumber = i + 1
|
|
report.Operations[i].Status = "skipped"
|
|
}
|
|
var runErr error
|
|
atomic := opts.AtomicExecution
|
|
if atomic && opts.ShouldContinue() {
|
|
runErr = errors.New("atomic execution cannot be combined with continue-on-error")
|
|
} else if atomic && (backend.Begin == nil || backend.Commit == nil || backend.Rollback == nil) {
|
|
runErr = errors.New("atomic execution is not supported by this executor")
|
|
} else if atomic {
|
|
runErr = backend.Begin(ctx)
|
|
}
|
|
if runErr == nil {
|
|
report.Outcome = "nontransactional"
|
|
if backend.ScriptManaged {
|
|
report.Outcome = "script_managed"
|
|
}
|
|
stopped := false
|
|
for i := range report.Operations {
|
|
op := &report.Operations[i]
|
|
if stopped || op.Skip || strings.TrimSpace(op.Statement) == "" {
|
|
continue
|
|
}
|
|
if err := backend.Exec(ctx, i, op.Statement); err != nil {
|
|
runErr = errors.Join(runErr, err)
|
|
op.Status, op.Error = "failed", err.Error()
|
|
report.FailedStatements++
|
|
report.Errors = append(report.Errors, ExecutionError{i + 1, op.Statement, err.Error()})
|
|
stopped = op.StopOnFailure || !opts.ShouldContinue()
|
|
} else {
|
|
op.Status = "applied"
|
|
if backend.ScriptManaged {
|
|
op.Status = "executed"
|
|
}
|
|
report.ExecutedStatements++
|
|
}
|
|
}
|
|
if atomic {
|
|
if report.FailedStatements > 0 {
|
|
// Cleanup uses a bounded context independent of caller cancellation.
|
|
cleanup, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
rollbackErr := backend.Rollback(cleanup)
|
|
cancel()
|
|
if rollbackErr == nil {
|
|
report.Outcome = "rolled_back"
|
|
for i := range report.Operations {
|
|
if report.Operations[i].Status == "applied" {
|
|
report.Operations[i].Status = "rolled_back"
|
|
report.RolledBackStatements++
|
|
}
|
|
}
|
|
report.ExecutedStatements = 0
|
|
} else {
|
|
report.Outcome = "unknown"
|
|
runErr = errors.Join(runErr, fmt.Errorf("rollback failed: %w", rollbackErr))
|
|
}
|
|
} else {
|
|
runErr = backend.Commit(ctx)
|
|
if runErr == nil {
|
|
report.Outcome = "committed"
|
|
} else {
|
|
report.Outcome = "unknown"
|
|
}
|
|
}
|
|
if report.Outcome == "rolled_back" || report.Outcome == "unknown" {
|
|
for i := range report.Schemas {
|
|
for j := range report.Schemas[i].Tables {
|
|
report.Schemas[i].Tables[j].Created = false
|
|
}
|
|
}
|
|
}
|
|
if report.Outcome == "unknown" {
|
|
for i := range report.Operations {
|
|
if report.Operations[i].Status == "applied" {
|
|
report.Operations[i].Status = "unknown"
|
|
}
|
|
}
|
|
report.ExecutedStatements = 0
|
|
}
|
|
}
|
|
}
|
|
for _, op := range report.Operations {
|
|
if op.Status == "skipped" {
|
|
report.SkippedStatements++
|
|
}
|
|
}
|
|
report.EndTime = time.Now().Format(time.RFC3339Nano)
|
|
if report.FailedStatements > 0 {
|
|
runErr = errors.Join(&ExecutionFailure{report}, runErr)
|
|
}
|
|
return FinishExecution(opts, report, runErr)
|
|
}
|
|
|
|
// FinishExecution publishes legacy counters and a typed report even on failure.
|
|
func FinishExecution(opts *WriterOptions, report *ExecutionReport, runErr error) error {
|
|
if opts.Metadata == nil {
|
|
opts.Metadata = map[string]interface{}{}
|
|
}
|
|
opts.Metadata["execution_report"] = report
|
|
opts.Metadata["execution_total"] = report.TotalStatements
|
|
opts.Metadata["execution_success"] = report.ExecutedStatements
|
|
opts.Metadata["execution_failed"] = report.FailedStatements
|
|
opts.Metadata["execution_skipped"] = report.SkippedStatements
|
|
opts.Metadata["execution_rolled_back"] = report.RolledBackStatements
|
|
if path, _ := opts.Metadata["report_path"].(string); path != "" {
|
|
if err := WriteExecutionReport(path, report); err != nil {
|
|
runErr = errors.Join(runErr, fmt.Errorf("failed to write requested execution report: %w", err))
|
|
}
|
|
}
|
|
return runErr
|
|
}
|
|
|
|
func WriteExecutionReport(path string, report *ExecutionReport) error {
|
|
file, err := os.Create(path)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to create report file: %w", err)
|
|
}
|
|
encoder := json.NewEncoder(file)
|
|
encoder.SetIndent("", " ")
|
|
return errors.Join(encoder.Encode(report), file.Close())
|
|
}
|
|
|
|
// StatementOperations creates a plan from the writers' structured statement lists.
|
|
func StatementOperations(statements []string) []ExecutionOperation {
|
|
ops := make([]ExecutionOperation, len(statements))
|
|
for i, stmt := range statements {
|
|
ops[i] = ExecutionOperation{Statement: stmt, StopOnFailure: strings.HasPrefix(strings.TrimSpace(stmt), "DO $relspec_preflight"), Skip: strings.TrimSpace(stmt) == "" || strings.HasPrefix(strings.TrimSpace(stmt), "--")}
|
|
}
|
|
return ops
|
|
}
|
|
|
|
// FailExecution records an invocation that failed before issuing any SQL.
|
|
func FailExecution(opts *WriterOptions, dialect string, operations []ExecutionOperation, cause error) error {
|
|
report := &ExecutionReport{Dialect: dialect, TotalStatements: len(operations), SkippedStatements: len(operations), Outcome: "not_started", Operations: append([]ExecutionOperation(nil), operations...)}
|
|
report.StartTime = time.Now().Format(time.RFC3339Nano)
|
|
report.EndTime = report.StartTime
|
|
for i := range report.Operations {
|
|
report.Operations[i].StatementNumber = i + 1
|
|
report.Operations[i].Status = "skipped"
|
|
}
|
|
return FinishExecution(opts, report, cause)
|
|
}
|