kandev/apps/backend/internal/task/repository/sqlite/workflow.go

package sqlite

import (
	"context"
	"database/sql"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/jmoiron/sqlx"

	"github.com/kandev/kandev/internal/db/dialect"
	"github.com/kandev/kandev/internal/task/models"
)

// AddTaskToWorkflow adds a task to a workflow with placement
func (r *Repository) AddTaskToWorkflow(ctx context.Context, taskID, workflowID, workflowStepID string, position int) error {
	_, err := r.db.ExecContext(ctx, r.db.Rebind(`
		UPDATE tasks SET workflow_id = ?, workflow_step_id = ?, position = ?, updated_at = ? WHERE id = ?
	`), workflowID, workflowStepID, position, time.Now().UTC(), taskID)
	return err
}

// RemoveTaskFromWorkflow removes a task from a workflow
func (r *Repository) RemoveTaskFromWorkflow(ctx context.Context, taskID, workflowID string) error {
	_, err := r.db.ExecContext(ctx, r.db.Rebind(`
		UPDATE tasks SET workflow_id = '', workflow_step_id = '', position = 0, updated_at = ? WHERE id = ? AND workflow_id = ?
	`), time.Now().UTC(), taskID, workflowID)
	return err
}

// Workflow operations

// CreateWorkflow creates a new workflow
func (r *Repository) CreateWorkflow(ctx context.Context, workflow *models.Workflow) error {
	r.prepareWorkflow(workflow)
	return r.insertWorkflow(ctx, r.db, workflow)
}

func (r *Repository) prepareWorkflow(workflow *models.Workflow) {
	if workflow.ID == "" {
		workflow.ID = uuid.New().String()
	}
	now := time.Now().UTC()
	workflow.CreatedAt = now
	workflow.UpdatedAt = now

}

func (r *Repository) insertWorkflow(ctx context.Context, exec sqlx.ExtContext, workflow *models.Workflow) error {
	// Auto-assign sort_order as max+1 within the workspace.
	// The caller's transaction keeps the read and insert coherent.
	var maxOrder int
	err := exec.QueryRowxContext(ctx, r.db.Rebind(
		`SELECT COALESCE(MAX(sort_order), -1) FROM workflows WHERE workspace_id = ?`,
	), workflow.WorkspaceID).Scan(&maxOrder)
	if err != nil {
		return fmt.Errorf("failed to get max sort_order: %w", err)
	}
	workflow.SortOrder = maxOrder + 1

	_, err = exec.ExecContext(ctx, r.db.Rebind(`
		INSERT INTO workflows (id, workspace_id, name, description, agent_profile_id, workflow_template_id, sort_order, hidden, style, source, source_path, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
	`), workflow.ID, workflow.WorkspaceID, workflow.Name, workflow.Description, workflow.AgentProfileID, workflow.WorkflowTemplateID, workflow.SortOrder, dialect.BoolToInt(workflow.Hidden), normalizeWorkflowStyle(workflow.Style), normalizeWorkflowSource(workflow.Source), workflow.SourcePath, workflow.CreatedAt, workflow.UpdatedAt)

	return err
}

// normalizeWorkflowSource returns a value fit for the workflows.source column.
// Empty or unknown sources collapse to the manual default.
func normalizeWorkflowSource(source string) string {
	switch source {
	case models.WorkflowSourceManual, models.WorkflowSourceGitHub:
		return source
	}
	return models.WorkflowSourceManual
}

// normalizeWorkflowStyle returns a value fit for the workflows.style column.
// Empty or unknown styles collapse to the kanban default so existing callers
// stay schema-compliant without having to set the field explicitly.
func normalizeWorkflowStyle(style string) string {
	switch style {
	case models.WorkflowStyleKanban, models.WorkflowStyleOffice, models.WorkflowStyleCustom:
		return style
	}
	return models.WorkflowStyleKanban
}

const workflowSelectColumns = `
	id, workspace_id, name, description, agent_profile_id,
	workflow_template_id, sort_order, hidden, style, source, source_path, created_at, updated_at
`

type workflowScanner interface {
	Scan(dest ...interface{}) error
}

func scanWorkflowRow(scanner workflowScanner) (*models.Workflow, error) {
	workflow := &models.Workflow{}
	var workflowTemplateID, agentProfileID, style, source, sourcePath sql.NullString
	var hidden int
	if err := scanner.Scan(
		&workflow.ID,
		&workflow.WorkspaceID,
		&workflow.Name,
		&workflow.Description,
		&agentProfileID,
		&workflowTemplateID,
		&workflow.SortOrder,
		&hidden,
		&style,
		&source,
		&sourcePath,
		&workflow.CreatedAt,
		&workflow.UpdatedAt,
	); err != nil {
		return nil, err
	}
	workflow.Hidden = hidden != 0
	if agentProfileID.Valid {
		workflow.AgentProfileID = agentProfileID.String
	}
	if workflowTemplateID.Valid {
		workflow.WorkflowTemplateID = &workflowTemplateID.String
	}
	if style.Valid && style.String != "" {
		workflow.Style = style.String
	} else {
		workflow.Style = models.WorkflowStyleKanban
	}
	workflow.Source = normalizeWorkflowSource(source.String)
	if sourcePath.Valid {
		workflow.SourcePath = sourcePath.String
	}
	return workflow, nil
}

func scanWorkflowRows(rows *sql.Rows) ([]*models.Workflow, error) {
	var result []*models.Workflow
	for rows.Next() {
		workflow, err := scanWorkflowRow(rows)
		if err != nil {
			return nil, err
		}
		result = append(result, workflow)
	}
	return result, rows.Err()
}

// GetWorkflow retrieves a workflow by ID
func (r *Repository) GetWorkflow(ctx context.Context, id string) (*models.Workflow, error) {
	workflow, err := scanWorkflowRow(r.ro.QueryRowContext(ctx, r.ro.Rebind(fmt.Sprintf(`
		SELECT %s
		FROM workflows WHERE id = ?
	`, workflowSelectColumns)), id))
	if err == sql.ErrNoRows {
		return nil, fmt.Errorf("workflow not found: %s", id)
	}
	if err != nil {
		return nil, err
	}
	return workflow, nil
}

// UpdateWorkflow updates an existing workflow
func (r *Repository) UpdateWorkflow(ctx context.Context, workflow *models.Workflow) error {
	workflow.UpdatedAt = time.Now().UTC()

	result, err := r.db.ExecContext(ctx, r.db.Rebind(`
		UPDATE workflows SET name = ?, description = ?, agent_profile_id = ?, workflow_template_id = ?, hidden = ?, style = ?, source = ?, source_path = ?, updated_at = ? WHERE id = ?
	`), workflow.Name, workflow.Description, workflow.AgentProfileID, workflow.WorkflowTemplateID, dialect.BoolToInt(workflow.Hidden), normalizeWorkflowStyle(workflow.Style), normalizeWorkflowSource(workflow.Source), workflow.SourcePath, workflow.UpdatedAt, workflow.ID)
	if err != nil {
		return err
	}

	rows, _ := result.RowsAffected()
	if rows == 0 {
		return fmt.Errorf("workflow not found: %s", workflow.ID)
	}
	return nil
}

// DeleteWorkflowsByWorkspace deletes all workflows for a workspace except the excluded IDs (E2E cleanup).
// Relies on CASCADE foreign keys to remove workflow_steps.
func (r *Repository) DeleteWorkflowsByWorkspace(ctx context.Context, workspaceID string, excludeIDs []string) (int64, error) {
	if len(excludeIDs) == 0 {
		result, err := r.db.ExecContext(ctx, r.db.Rebind(`DELETE FROM workflows WHERE workspace_id = ?`), workspaceID)
		if err != nil {
			return 0, err
		}
		rows, _ := result.RowsAffected()
		return rows, nil
	}

	query, args, err := sqlx.In(`DELETE FROM workflows WHERE workspace_id = ? AND id NOT IN (?)`, workspaceID, excludeIDs)
	if err != nil {
		return 0, err
	}
	result, err := r.db.ExecContext(ctx, r.db.Rebind(query), args...)
	if err != nil {
		return 0, err
	}
	rows, _ := result.RowsAffected()
	return rows, nil
}

// DeleteWorkflow deletes a workflow by ID
func (r *Repository) DeleteWorkflow(ctx context.Context, id string) error {
	result, err := r.db.ExecContext(ctx, r.db.Rebind(`DELETE FROM workflows WHERE id = ?`), id)
	if err != nil {
		return err
	}

	rows, _ := result.RowsAffected()
	if rows == 0 {
		return fmt.Errorf("workflow not found: %s", id)
	}
	return nil
}

// ListWorkflows returns workflows for the given workspace, excluding hidden by default.
// Pass includeHidden=true to also return system-only workflows like Improve Kandev.
func (r *Repository) ListWorkflows(ctx context.Context, workspaceID string, includeHidden bool) ([]*models.Workflow, error) {
	query := fmt.Sprintf(`
		SELECT %s
		FROM workflows
	`, workflowSelectColumns)
	var args []interface{}
	var conditions []string
	if workspaceID != "" {
		conditions = append(conditions, "workspace_id = ?")
		args = append(args, workspaceID)
	}
	if !includeHidden {
		conditions = append(conditions, "hidden = 0")
	}
	for i, c := range conditions {
		if i == 0 {
			query += " WHERE " + c
		} else {
			query += " AND " + c
		}
	}
	query += " ORDER BY sort_order ASC, created_at ASC"

	rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query), args...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	return scanWorkflowRows(rows)
}

// ReorderWorkflows updates sort_order for workflows within a workspace using a transaction.
func (r *Repository) ReorderWorkflows(ctx context.Context, workspaceID string, workflowIDs []string) error {
	tx, err := r.db.BeginTxx(ctx, nil)
	if err != nil {
		return fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	now := time.Now().UTC()
	for i, id := range workflowIDs {
		result, err := tx.ExecContext(ctx, r.db.Rebind(
			`UPDATE workflows SET sort_order = ?, updated_at = ? WHERE id = ? AND workspace_id = ?`,
		), i, now, id, workspaceID)
		if err != nil {
			return fmt.Errorf("failed to update sort_order for workflow %s: %w", id, err)
		}
		rows, _ := result.RowsAffected()
		if rows == 0 {
			return fmt.Errorf("workflow not found in workspace: %s", id)
		}
	}
	return tx.Commit()
}