Add PostgreSQL plugin

internal/pkg/object/command/postgres/postgres.go

@@ -0,0 +1,140 @@
+package postgres
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/patterninc/heimdall/internal/pkg/database"
+	pkgcontext "github.com/patterninc/heimdall/pkg/context"
+	"github.com/patterninc/heimdall/pkg/object/cluster"
+	"github.com/patterninc/heimdall/pkg/object/job"
+	"github.com/patterninc/heimdall/pkg/plugin"
+	"github.com/patterninc/heimdall/pkg/result"
+	"github.com/patterninc/heimdall/pkg/result/column"
+)
+
+// postgresJobContext represents the context for a PostgreSQL job
+type postgresJobContext struct {
+	Query        string `yaml:"query,omitempty" json:"query,omitempty"`
+	ReturnResult bool   `yaml:"return_result,omitempty" json:"return_result,omitempty"`
+}
+
+type postgresClusterContext struct {
+	ConnectionString string `yaml:"connection_string,omitempty" json:"connection_string,omitempty"`
+}
+
+type postgresCommandContext struct {
+	mu sync.Mutex
+}
+
+// New creates a new PostgreSQL plugin handler.
+func New(_ *pkgcontext.Context) (plugin.Handler, error) {
+	p := &postgresCommandContext{}
+	return p.handler, nil
+}
+
+// Handler for the PostgreSQL query execution.
+func (p *postgresCommandContext) handler(r *plugin.Runtime, j *job.Job, c *cluster.Cluster) error {
+	jobContext := &postgresJobContext{}
+	if j.Context != nil {
+		if err := j.Context.Unmarshal(jobContext); err != nil {
+			return fmt.Errorf("failed to unmarshal job context: %w", err)
+		}
+	}
+	if jobContext.Query == "" {
+		return fmt.Errorf("query is required in job context")
+	}
+
+	clusterContext := &postgresClusterContext{}
+	if c.Context != nil {
+		if err := c.Context.Unmarshal(clusterContext); err != nil {
+			return fmt.Errorf("failed to unmarshal cluster context: %w", err)
+		}
+	}
+	if clusterContext.ConnectionString == "" {
+		return fmt.Errorf("connection_string is required in cluster context")
+	}
+
+	db := &database.Database{ConnectionString: clusterContext.ConnectionString}
+
+	if jobContext.ReturnResult {
+		return executeSyncQuery(db, jobContext.Query, j)
+	}
+	return p.executeAsyncQueries(db, jobContext.Query, j)
+}
+
+func executeSyncQuery(db *database.Database, query string, j *job.Job) error {
+	// Allow a single query, even if it ends with a semicolon
+	queries := splitAndTrimQueries(query)
+	if len(queries) != 1 {
+		return fmt.Errorf("multiple queries are not allowed when return_result is true")
+	}
+
+	sess, err := db.NewSession(false)
+	if err != nil {
+		return fmt.Errorf("failed to open PostgreSQL connection: %w", err)
+	}
+	defer sess.Close()
+
+	rows, err := sess.Query(queries[0])
+	if err != nil {
+		return fmt.Errorf("PostgreSQL query execution failed: %w", err)
+	}
+	defer rows.Close()
+
+	rowsResult, err := result.FromRows(rows)
+	if err != nil {
+		return fmt.Errorf("failed to process PostgreSQL query results: %w", err)
+	}
+
+	j.Result = rowsResult
+	return nil
+}
+
+func (p *postgresCommandContext) executeAsyncQueries(db *database.Database, query string, j *job.Job) error {
+	sess, err := db.NewSession(false)
+	if err != nil {
+		j.Result = &result.Result{
+			Columns: []*column.Column{{
+				Name: "error",
+				Type: column.Type("string"),
+			}},
+			Data: [][]any{{fmt.Sprintf("Async PostgreSQL connection error: %v", err)}},
+		}
+		return fmt.Errorf("Async PostgreSQL connection error: %v", err)
+	}
+	defer sess.Close()
+
+	_, err = sess.Exec(query)
+	if err != nil {
+		j.Result = &result.Result{
+			Columns: []*column.Column{{
+				Name: "error",
+				Type: column.Type("string"),
+			}},
+			Data: [][]any{{fmt.Sprintf("Async PostgreSQL query execution error: %v", err)}},
+		}
+		return fmt.Errorf("Async PostgreSQL query execution error: %v", err)
+	}
+
+	j.Result = &result.Result{
+		Columns: []*column.Column{{
+			Name: "message",
+			Type: column.Type("string"),
+		}},
+		Data: [][]any{{"All queries executed successfully"}},
+	}
+	return nil
+}
+
+func splitAndTrimQueries(query string) []string {
+	queries := []string{}
+	for _, q := range strings.Split(query, ";") {
+		q = strings.TrimSpace(q)
+		if q != "" {
+			queries = append(queries, q)
+		}
+	}
+	return queries
+}

plugins/postgres/README.md

@@ -0,0 +1,73 @@
+# PostgreSQL Plugin
+
+This plugin provides an interface to PostgreSQL databases with support for direct SQL queries, SQL files, and batch execution.
+
+## Features
+
+- Execute single or multiple SQL statements
+- Support for both direct query strings and SQL file execution
+- Synchronous (return_result: true) and asynchronous (return_result: false) execution modes
+- Error reporting with line number and query for batch execution
+- Transaction support for batch execution
+
+## Configuration
+
+### Cluster Context
+
+```yaml
+connection_string: postgresql://user:password@host:port/database # Required
+```
+
+### Job Context
+
+```yaml
+query: SELECT * FROM my_table # Required - SQL query to execute or path to .sql file
+return_result: true # Optional - Whether to return query results (default: false)
+```
+
+#### Execution Modes
+
+1. **Return Results Mode** (`return_result: true`):
+   - Only a single query is allowed (trailing semicolon is fine)
+   - Returns query results as structured data
+   - Fails if multiple queries are provided
+
+2. **Execute Only Mode** (`return_result: false`):
+   - Multiple queries separated by `;` are allowed
+   - Executes all queries in order, stops and fails on first error
+   - Returns error message with line number and query if any query fails
+   - Returns a success message if all queries succeed
+
+## Usage
+
+```yaml
+# Example 1: SELECT query with results
+- name: postgresql-select
+  description: Execute a SELECT query and return results
+  command: postgres
+  cluster: postgres-cluster
+  context:
+    query: SELECT * FROM my_table WHERE status = 'active'
+    return_result: true
+
+# Example 2: Execute an UPDATE statement
+- name: postgresql-update
+  description: Execute an UPDATE statement
+  command: postgres
+  cluster: postgres-cluster
+  context:
+    query: UPDATE my_table SET status = 'inactive' WHERE last_active < '2025-01-01'
+    return_result: false
+
+
+# Example 3: Execute multiple statements in a batch
+- name: postgresql-batch
+  description: Execute multiple SQL statements in a batch
+  command: postgres
+  cluster: postgres-cluster
+  context:
+    query: |
+      INSERT INTO orders (id, customer_id, amount) VALUES ('ORD001', 'CUST123', 299.99);
+      UPDATE inventory SET stock = stock - 1 WHERE product_id = 'PROD456';
+    return_result: false
+```

plugins/postgres/postgres.go

@@ -0,0 +1,12 @@
+package main
+
+import (
+	"github.com/patterninc/heimdall/internal/pkg/object/command/postgres"
+	"github.com/patterninc/heimdall/pkg/context"
+	"github.com/patterninc/heimdall/pkg/plugin"
+)
+
+// New creates a new instance of the postgres plugin.
+func New(c *context.Context) (plugin.Handler, error) {
+	return postgres.New(c)
+}