+ {{if .HistoryMode}}
+
+
+
+ Matching attempts: {{.FilteredCount}} (all dates)
+
+
+ {{else}}
{{if .HourlyStats}}
+ {{end}}
{{if gt .TotalPages 1}}
@@ -58,11 +67,11 @@
@@ -192,7 +201,11 @@
No tasks found
+ {{if .HistoryMode}}
+ No tasks were found for the selected ID(s) within retention.
+ {{else}}
No tasks were found for {{.Date}} with the current filters.
+ {{end}}
|
{{end}}
@@ -200,7 +213,9 @@
- {{if or .Filter.ID .Filter.Type .Filter.Job .Filter.Result}}
+ {{if .HistoryMode}}
+ Showing: {{len .Tasks}} | Matching Attempts: {{.FilteredCount}}
+ {{else if or .Filter.ID .Filter.Type .Filter.Job .Filter.Result}}
Showing: {{len .Tasks}} | Filtered Tasks: {{.FilteredCount}} | Total Tasks: {{.Counts.Total}}
{{else}}
Showing: {{len .Tasks}} | Total Tasks: {{.Counts.Total}}
@@ -212,11 +227,11 @@
@@ -259,8 +274,8 @@
window.FlowlordTask.init(config);
}
- // Render hourly task chart
- {{if .HourlyStats}}
+ // Render hourly task chart (day view only)
+ {{if and .HourlyStats (not .HistoryMode)}}
renderHourlyChart();
// Re-render chart on window resize with debouncing
diff --git a/apps/flowlord/handler_test.go b/apps/flowlord/handler_test.go
index 56b4841..43308ef 100644
--- a/apps/flowlord/handler_test.go
+++ b/apps/flowlord/handler_test.go
@@ -428,6 +428,14 @@ func TestAlertHTML(t *testing.T) {
datesWithData := []string{"2024-01-14", "2024-01-15", "2024-01-16"}
htmlContent := alertHTML(sampleAlerts, trial.TimeDay("2024-01-15"), datesWithData)
+ html := string(htmlContent)
+ if !strings.Contains(html, `class="task-id-link" href="./task_preview.html?id=task-001"`) {
+ t.Error("alert task link should use id-only history URL")
+ }
+ if strings.Contains(html, `class="task-id-link" href="./task_preview.html?date=`) {
+ t.Error("alert task link should not include date from task_time")
+ }
+
// Validate HTML using the new function
if err := validateHTML(htmlContent); err != nil {
t.Errorf("HTML validation failed: %v", err)
@@ -548,7 +556,7 @@ func TestTaskHTML(t *testing.T) {
// Pass sample dates with data for calendar highlighting
datesWithData := []string{"2024-01-15"}
_, hourlyStats := taskStats.HourlyCounts(filter)
- html := taskHTML(testTasks, taskStats, len(testTasks), date, filter, datesWithData, 0, hourlyStats)
+ html := taskHTML(testTasks, taskStats, len(testTasks), date, filter, datesWithData, 0, hourlyStats, false)
// Validate HTML using the new function
if err := validateHTML(html); err != nil {
@@ -857,3 +865,41 @@ func TestRerunProducerFailure(t *testing.T) {
t.Fatalf("rerun attempts after failure = %d, want 1", newAttempts)
}
}
+
+func TestHTMLTask_historyModeFilterOptions(t *testing.T) {
+ sqlDB := &sqlite.SQLite{LocalPath: ":memory:"}
+ if err := sqlDB.Open(testPath+"/workflow/f3.toml", nil); err != nil {
+ t.Fatal(err)
+ }
+ sqlDB.Add(task.Task{
+ ID: "pipe-1", Type: "alpha", Job: "load",
+ Created: "2024-01-15T10:00:00Z", Result: task.CompleteResult,
+ })
+ sqlDB.Add(task.Task{
+ ID: "pipe-1", Type: "zebra", Job: "check",
+ Created: "2024-01-16T09:00:00Z", Result: task.ErrResult,
+ })
+
+ tm := &taskMaster{taskCache: sqlDB}
+ r := httptest.NewRequest(http.MethodGet, "/web/task?id=pipe-1", nil)
+ w := httptest.NewRecorder()
+ tm.htmlTask(w, r)
+
+ body := w.Body.String()
+ if w.Code != http.StatusOK {
+ t.Fatalf("status = %d", w.Code)
+ }
+ if !strings.Contains(body, `taskTypes: ["alpha","zebra"]`) {
+ t.Errorf("history mode column filters missing type options; got taskTypes snippet from body")
+ if idx := strings.Index(body, "taskTypes:"); idx >= 0 {
+ end := idx + 80
+ if end > len(body) {
+ end = len(body)
+ }
+ t.Logf("snippet: %s", body[idx:end])
+ }
+ }
+ if !strings.Contains(body, `"load"`) || !strings.Contains(body, `"check"`) {
+ t.Error("history mode column filters missing job options")
+ }
+}
diff --git a/apps/flowlord/sqlite/alerts.go b/apps/flowlord/sqlite/alerts.go
index 8336f33..efe3b45 100644
--- a/apps/flowlord/sqlite/alerts.go
+++ b/apps/flowlord/sqlite/alerts.go
@@ -45,16 +45,13 @@ func (s *SQLite) AddAlert(t task.Task, message string) error {
taskID = "unknown"
}
- // Extract job using helper function
- job := extractJobFromTask(t)
-
// Get task time using tmpl.TaskTime function
taskTime := tmpl.TaskTime(t)
_, err := s.db.Exec(`
INSERT INTO alert_records (task_id, task_time, task_type, job, msg)
VALUES (?, ?, ?, ?, ?)
- `, taskID, taskTime, t.Type, job, message)
+ `, taskID, taskTime, t.Type, t.Job, message)
if err != nil {
return err
diff --git a/apps/flowlord/sqlite/dates.go b/apps/flowlord/sqlite/dates.go
index 6c413a5..93dd5da 100644
--- a/apps/flowlord/sqlite/dates.go
+++ b/apps/flowlord/sqlite/dates.go
@@ -49,50 +49,6 @@ func (s *SQLite) updateDateIndex(timestamp, dataType string) {
}
}
-// GetDatesWithData returns a list of dates (YYYY-MM-DD format) that have any data
-// for tasks, alerts, or files within the retention period
-func (s *SQLite) GetDatesWithData() ([]string, error) {
- s.mu.Lock()
- defer s.mu.Unlock()
-
- query := `
- SELECT DISTINCT date_val FROM (
- SELECT DISTINCT DATE(created) AS date_val FROM task_records
- WHERE created >= datetime('now', '-' || ? || ' days')
- UNION
- SELECT DISTINCT DATE(created_at) AS date_val FROM alert_records
- WHERE created_at >= datetime('now', '-' || ? || ' days')
- UNION
- SELECT DISTINCT DATE(received_at) AS date_val FROM file_messages
- WHERE received_at >= datetime('now', '-' || ? || ' days')
- )
- ORDER BY date_val DESC
- `
-
- // Use retention period in days (default 90)
- retentionDays := int(s.Retention.Hours() / 24)
- if retentionDays == 0 {
- retentionDays = 90
- }
-
- rows, err := s.db.Query(query, retentionDays, retentionDays, retentionDays)
- if err != nil {
- return nil, err
- }
- defer rows.Close()
-
- var dates []string
- for rows.Next() {
- var date string
- if err := rows.Scan(&date); err != nil {
- continue
- }
- dates = append(dates, date)
- }
-
- return dates, nil
-}
-
// DatesByType returns a list of dates (YYYY-MM-DD format) that have data for the specified type
// dataType can be "tasks", "alerts", or "files"
// This uses the date_index table for instant lookups
@@ -137,21 +93,6 @@ func (s *SQLite) DatesByType(dataType string) ([]string, error) {
return dates, nil
}
-// GetDatesWithTasks returns a list of dates (YYYY-MM-DD format) that have task records
-func (s *SQLite) GetDatesWithTasks() ([]string, error) {
- return s.DatesByType("tasks")
-}
-
-// GetDatesWithAlerts returns a list of dates (YYYY-MM-DD format) that have alert records
-func (s *SQLite) GetDatesWithAlerts() ([]string, error) {
- return s.DatesByType("alerts")
-}
-
-// GetDatesWithFiles returns a list of dates (YYYY-MM-DD format) that have file message records
-func (s *SQLite) GetDatesWithFiles() ([]string, error) {
- return s.DatesByType("files")
-}
-
// RebuildDateIndex scans all tables and rebuilds the date_index table
// This should be called once during migration or can be exposed as an admin endpoint
func (s *SQLite) RebuildDateIndex() error {
diff --git a/apps/flowlord/sqlite/files.go b/apps/flowlord/sqlite/files.go
index b61a6e1..2608339 100644
--- a/apps/flowlord/sqlite/files.go
+++ b/apps/flowlord/sqlite/files.go
@@ -21,21 +21,6 @@ type FileMessage struct {
TaskNames []string `json:"task_names"`
}
-// FileMessageWithTasks represents a file message with associated task details
-type FileMessageWithTasks struct {
- FileID int `json:"file_id"`
- Path string `json:"path"`
- TaskTime time.Time `json:"task_time"`
- ReceivedAt time.Time `json:"received_at"`
- TaskID string `json:"task_id"`
- TaskType string `json:"task_type"`
- TaskJob string `json:"task_job"`
- TaskResult string `json:"task_result"`
- TaskCreated time.Time `json:"task_created"`
- TaskStarted time.Time `json:"task_started"`
- TaskEnded time.Time `json:"task_ended"`
-}
-
// AddFileMessage stores a file message in the database
func (s *SQLite) AddFileMessage(sts stat.Stats, taskIDs []string, taskNames []string) error {
s.mu.Lock()
@@ -78,51 +63,6 @@ func (s *SQLite) AddFileMessage(sts stat.Stats, taskIDs []string, taskNames []st
return nil
}
-// GetFileMessages retrieves file messages with optional filtering
-func (s *SQLite) GetFileMessages(limit int, offset int) ([]FileMessage, error) {
- s.mu.Lock()
- defer s.mu.Unlock()
-
- query := `
- SELECT id, path, SIZE, last_modified, received_at, task_time, task_ids, task_names
- FROM file_messages
- ORDER BY received_at DESC
- LIMIT ? OFFSET ?
- `
-
- rows, err := s.db.Query(query, limit, offset)
- if err != nil {
- return nil, err
- }
- defer rows.Close()
-
- var messages []FileMessage
- for rows.Next() {
- var msg FileMessage
- var taskIDsJSON, taskNamesJSON sql.NullString
-
- err := rows.Scan(
- &msg.ID, &msg.Path, &msg.Size, &msg.LastModified, &msg.ReceivedAt,
- &msg.TaskTime, &taskIDsJSON, &taskNamesJSON,
- )
- if err != nil {
- continue
- }
-
- // Parse JSON arrays
- if taskIDsJSON.Valid {
- json.Unmarshal([]byte(taskIDsJSON.String), &msg.TaskIDs)
- }
- if taskNamesJSON.Valid {
- json.Unmarshal([]byte(taskNamesJSON.String), &msg.TaskNames)
- }
-
- messages = append(messages, msg)
- }
-
- return messages, nil
-}
-
// GetFileMessagesByDate retrieves file messages for a specific date
func (s *SQLite) GetFileMessagesByDate(date time.Time) ([]FileMessage, error) {
s.mu.Lock()
@@ -169,65 +109,3 @@ func (s *SQLite) GetFileMessagesByDate(date time.Time) ([]FileMessage, error) {
return messages, nil
}
-// GetFileMessagesWithTasks retrieves file messages with their associated task details
-func (s *SQLite) GetFileMessagesWithTasks(limit int, offset int) ([]FileMessageWithTasks, error) {
- s.mu.Lock()
- defer s.mu.Unlock()
-
- query := `
- SELECT
- fm.id, fm.path, fm.task_time, fm.received_at,
- json_extract(t.value, '$') AS task_id,
- tl.type AS task_type,
- tl.job AS task_job,
- tl.result AS task_result,
- tl.created AS task_created,
- tl.started AS task_started,
- tl.ended AS task_ended
- FROM file_messages fm,
- json_each(fm.task_ids) AS t
- JOIN task_log tl ON json_extract(t.value, '$') = tl.id
- WHERE fm.task_ids IS NOT NULL
- ORDER BY fm.received_at DESC, fm.id
- LIMIT ? OFFSET ?
- `
-
- rows, err := s.db.Query(query, limit, offset)
- if err != nil {
- return nil, err
- }
- defer rows.Close()
-
- var results []FileMessageWithTasks
- for rows.Next() {
- var result FileMessageWithTasks
- var taskCreated, taskStarted, taskEnded sql.NullString
-
- err := rows.Scan(
- &result.FileID, &result.Path, &result.TaskTime, &result.ReceivedAt,
- &result.TaskID, &result.TaskType, &result.TaskJob, &result.TaskResult,
- &taskCreated, &taskStarted, &taskEnded,
- )
- if err != nil {
- continue
- }
-
- // Parse timestamps
- if taskCreated.Valid {
- result.TaskCreated, _ = time.Parse(time.RFC3339, taskCreated.String)
- }
- if taskStarted.Valid {
- result.TaskStarted, _ = time.Parse(time.RFC3339, taskStarted.String)
- }
- if taskEnded.Valid {
- result.TaskEnded, _ = time.Parse(time.RFC3339, taskEnded.String)
- }
-
- results = append(results, result)
- }
-
- return results, nil
-}
-
-
-
diff --git a/apps/flowlord/sqlite/filter.go b/apps/flowlord/sqlite/filter.go
index ca609d2..ad95c11 100644
--- a/apps/flowlord/sqlite/filter.go
+++ b/apps/flowlord/sqlite/filter.go
@@ -19,15 +19,17 @@ const (
// Empty / nil slice fields are ignored in the query.
// Type, Job, and Result support multi-select via comma-separated or repeated query params
// (parsed with github.com/jbsmith7741/uri).
+// Date is set by the handler (not from URI unmarshal): nil means no day scope (ID history).
type TaskFilter struct {
- ID []string `uri:"id"`
- Type []string `uri:"type"`
- Job []string `uri:"job"`
- Result []string `uri:"result"` // complete, error, alert, warn, or running
- Sort string `uri:"sort"`
- Direction string `uri:"direction"` // "asc" or "desc"
- Page int `uri:"page"` // 1-based, default: 1
- Limit int `uri:"-"` // not taken from the URL
+ ID []string `uri:"id"`
+ Type []string `uri:"type"`
+ Job []string `uri:"job"`
+ Result []string `uri:"result"` // complete, error, alert, warn, or running
+ Sort string `uri:"sort"`
+ Direction string `uri:"direction"` // "asc" or "desc"
+ Page int `uri:"page"` // 1-based, default: 1
+ Limit int `uri:"-"` // not taken from the URL
+ Date *time.Time `uri:"-"` // nil = all dates; set = DATE(created) day scope
}
// taskSortColumns is the set of safe column names allowed in ORDER BY.
@@ -97,13 +99,17 @@ func (f *TaskFilter) orderByClause() string {
return "ORDER BY " + f.Sort + " " + dir
}
-// whereForDate builds a WHERE clause for tasks on a given date with optional ID/type/job/result filters.
-func (f *TaskFilter) whereForDate(date time.Time) (string, []any) {
+// whereForFilter builds a WHERE clause from the filter fields.
+// When Date is non-nil, results are scoped to DATE(created) = that day (day view).
+// When Date is nil, no date constraint is applied (ID history view).
+func (f *TaskFilter) whereForFilter() (string, []any) {
if f == nil {
f = &TaskFilter{}
}
w := &whereBuilder{}
- w.And("DATE(created) = ?", date.Format("2006-01-02"))
+ if f.Date != nil {
+ w.And("DATE(created) = ?", f.Date.Format("2006-01-02"))
+ }
w.In("id", f.ID)
w.In("type", f.Type)
w.In("job", f.Job)
diff --git a/apps/flowlord/sqlite/filter_test.go b/apps/flowlord/sqlite/filter_test.go
index 40ea5f4..0a0925f 100644
--- a/apps/flowlord/sqlite/filter_test.go
+++ b/apps/flowlord/sqlite/filter_test.go
@@ -2,6 +2,7 @@ package sqlite
import (
"testing"
+ "time"
"github.com/hydronica/trial"
"github.com/jbsmith7741/uri"
@@ -192,3 +193,46 @@ func TestWhereBuilder(t *testing.T) {
}
trial.New(fn, cases).SubTest(t)
}
+
+func TestTaskFilter_whereForFilter(t *testing.T) {
+ day := time.Date(2024, 1, 15, 0, 0, 0, 0, time.UTC)
+ type output struct {
+ SQL string
+ Args []any
+ }
+ fn := func(filter TaskFilter) (output, error) {
+ sql, args := filter.whereForFilter()
+ return output{SQL: sql, Args: args}, nil
+ }
+ cases := trial.Cases[TaskFilter, output]{
+ "day scope with date": {
+ Input: TaskFilter{Date: &day},
+ Expected: output{SQL: "WHERE DATE(created) = ?", Args: []any{"2024-01-15"}},
+ },
+ "nil date omits date clause": {
+ Input: TaskFilter{ID: []string{"pipe-1"}},
+ Expected: output{
+ SQL: "WHERE id IN (?)",
+ Args: []any{"pipe-1"},
+ },
+ },
+ "nil date with type still omits date": {
+ Input: TaskFilter{
+ ID: []string{"pipe-1"},
+ Type: []string{"alpha"},
+ },
+ Expected: output{
+ SQL: "WHERE id IN (?) AND type IN (?)",
+ Args: []any{"pipe-1", "alpha"},
+ },
+ },
+ "day scope with id still includes date": {
+ Input: TaskFilter{Date: &day, ID: []string{"pipe-1"}},
+ Expected: output{
+ SQL: "WHERE DATE(created) = ? AND id IN (?)",
+ Args: []any{"2024-01-15", "pipe-1"},
+ },
+ },
+ }
+ trial.New(fn, cases).SubTest(t)
+}
diff --git a/apps/flowlord/sqlite/schema.sql b/apps/flowlord/sqlite/schema.sql
index 92e6f14..01b84d1 100644
--- a/apps/flowlord/sqlite/schema.sql
+++ b/apps/flowlord/sqlite/schema.sql
@@ -25,6 +25,7 @@ CREATE INDEX IF NOT EXISTS idx_task_records_job ON task_records (job);
CREATE INDEX IF NOT EXISTS idx_task_records_created ON task_records (created);
CREATE INDEX IF NOT EXISTS idx_task_records_type_job ON task_records (type, job);
CREATE INDEX IF NOT EXISTS idx_task_records_date_range ON task_records (created, ended);
+CREATE INDEX IF NOT EXISTS idx_task_records_id_created ON task_records (id, created);
-- Create a view that calculates task and queue times
DROP VIEW IF EXISTS tasks;
diff --git a/apps/flowlord/sqlite/sqlite.go b/apps/flowlord/sqlite/sqlite.go
index 3f062a1..56949b3 100644
--- a/apps/flowlord/sqlite/sqlite.go
+++ b/apps/flowlord/sqlite/sqlite.go
@@ -27,7 +27,8 @@ const (
// Version 1: Initial schema
// Version 2: Added date_index table for performance optimization
// Version 3: Store 'running' for in-progress tasks (was empty string)
-const currentSchemaVersion = 3
+// Version 4: Index on (id, created) for cross-date ID history lookups
+const currentSchemaVersion = 4
type SQLite struct {
LocalPath string
@@ -247,14 +248,20 @@ func (o *SQLite) migrateSchema(currentVersion int) error {
log.Println("Successfully migrated to schema version 3")
}
- // Add future migrations here as needed:
- // Example:
- // if currentVersion < 3 {
- // db := o.db
- // // Drop column by recreating table (since data loss is OK)
- // db.Exec("DROP TABLE IF EXISTS task_records")
- // // schema.sql will recreate it with correct structure
- // }
+ // Version 3 → 4: Index on (id, created) for cross-date ID history
+ if currentVersion < 4 {
+ log.Println("Migrating schema from version 3 to 4 (adding id,created index)")
+
+ _, err := o.db.Exec(`
+ CREATE INDEX IF NOT EXISTS idx_task_records_id_created
+ ON task_records (id, created)
+ `)
+ if err != nil {
+ return fmt.Errorf("failed to create idx_task_records_id_created: %w", err)
+ }
+
+ log.Println("Successfully migrated to schema version 4")
+ }
return nil
}
diff --git a/apps/flowlord/sqlite/sqlite_test.go b/apps/flowlord/sqlite/sqlite_test.go
index 4213343..c1d6d73 100644
--- a/apps/flowlord/sqlite/sqlite_test.go
+++ b/apps/flowlord/sqlite/sqlite_test.go
@@ -94,22 +94,6 @@ func TestDatesByType(t *testing.T) {
if err == nil {
t.Error("Expected error for invalid data type, got nil")
}
-
- // Test backward compatibility methods
- taskDates2, _ := db.GetDatesWithTasks()
- if len(taskDates2) != len(taskDates) {
- t.Error("GetDatesWithTasks() should return same results as DatesByType('tasks')")
- }
-
- alertDates2, _ := db.GetDatesWithAlerts()
- if len(alertDates2) != len(alertDates) {
- t.Error("GetDatesWithAlerts() should return same results as DatesByType('alerts')")
- }
-
- fileDates2, _ := db.GetDatesWithFiles()
- if len(fileDates2) != len(fileDates) {
- t.Error("GetDatesWithFiles() should return same results as DatesByType('files')")
- }
}
func TestMigrationNormalizesEmptyResult(t *testing.T) {
diff --git a/apps/flowlord/sqlite/stats.go b/apps/flowlord/sqlite/stats.go
index 30b75e5..fc96852 100644
--- a/apps/flowlord/sqlite/stats.go
+++ b/apps/flowlord/sqlite/stats.go
@@ -260,7 +260,7 @@ func (ts TaskStats) TotalCounts() TaskCounts {
}
// HourlyCounts returns total and hourly counts with optional filtering by type, job, and result.
-// ID filtering is not supported here; use SQLite.GetHourlyCountsByDate when filter.ID is set.
+// ID filtering is not applied here; the tasks page omits the hourly chart in ID history mode.
// The hourly array contains 24 TaskCounts where index represents the hour (0-23).
func (ts TaskStats) HourlyCounts(filter *TaskFilter) (TaskCounts, [24]TaskCounts) {
var total TaskCounts
@@ -304,11 +304,3 @@ func addTimesToHourly(times []time.Time, result string, filter *TaskFilter, tota
incrementHourly(result, t.Hour(), total, hourly)
}
}
-
-func addTaskHourlyCounts(tsk task.Task, filter *TaskFilter, total *TaskCounts, hourly *[24]TaskCounts) {
- result := string(tsk.Result)
- if !resultAllowed(result, filter) {
- return
- }
- incrementHourly(result, tmpl.TaskTime(tsk).Hour(), total, hourly)
-}
diff --git a/apps/flowlord/sqlite/tasks.go b/apps/flowlord/sqlite/tasks.go
index dfd46e9..819325d 100644
--- a/apps/flowlord/sqlite/tasks.go
+++ b/apps/flowlord/sqlite/tasks.go
@@ -4,7 +4,6 @@ import (
"database/sql"
"fmt"
"log"
- "net/url"
"strings"
"time"
@@ -12,6 +11,7 @@ import (
"github.com/pcelvng/task/bus"
"github.com/pcelvng/task-tools/tmpl"
+ "github.com/pcelvng/task-tools/workflow"
)
// TaskJob describes info about completed tasks that are within the cache
@@ -44,6 +44,7 @@ func (s *SQLite) Add(t task.Task) {
if t.ID == "" {
return
}
+ workflow.NormalizeJob(&t)
if t.Result == "" {
t.Result = ResultRunning
}
@@ -346,12 +347,12 @@ func (s *SQLite) SendFunc(p bus.Producer) func(string, *task.Task) error {
}
}
-// GetTasksByDate retrieves tasks for a specific date with optional filtering and pagination
-func (s *SQLite) GetTasksByDate(date time.Time, filter *TaskFilter) ([]TaskView, int, error) {
+// GetTasks retrieves tasks with optional filtering and pagination.
+// Set filter.Date for day view; leave Date nil for ID history across retained days.
+func (s *SQLite) GetTasks(filter *TaskFilter) ([]TaskView, int, error) {
s.mu.Lock()
defer s.mu.Unlock()
- // Handle nil filter or set defaults
if filter == nil {
filter = &TaskFilter{}
}
@@ -360,12 +361,11 @@ func (s *SQLite) GetTasksByDate(date time.Time, filter *TaskFilter) ([]TaskView,
filter.Limit = DefaultPageSize
}
if filter.Page <= 0 {
- filter.Page = 1 // default to first page
+ filter.Page = 1
}
- whereClause, args := filter.whereForDate(date)
+ whereClause, args := filter.whereForFilter()
- // Get total count of filtered results
countQuery := "SELECT COUNT(*) FROM tasks " + whereClause
var totalCount int
err := s.db.QueryRow(countQuery, args...).Scan(&totalCount)
@@ -373,13 +373,11 @@ func (s *SQLite) GetTasksByDate(date time.Time, filter *TaskFilter) ([]TaskView,
return nil, 0, err
}
- // Build main query with pagination
query := `SELECT id, type, job, info, result, meta, msg, task_seconds, task_time, queue_seconds, queue_time, created, started, ended
FROM tasks ` + whereClause + `
` + filter.orderByClause() + `
LIMIT ? OFFSET ?`
- // Calculate offset from page number
offset := (filter.Page - 1) * filter.Limit
args = append(args, filter.Limit, offset)
@@ -407,42 +405,39 @@ func (s *SQLite) GetTasksByDate(date time.Time, filter *TaskFilter) ([]TaskView,
return tasks, totalCount, nil
}
-// GetHourlyCountsByDate returns hourly task counts for a date, querying tasks directly so ID
-// filters (and all other filter fields) are applied per task.
-func (s *SQLite) GetHourlyCountsByDate(date time.Time, filter *TaskFilter) (TaskCounts, [24]TaskCounts, error) {
+// TypeJobKeys returns TaskStats keyed by distinct type:job matching the filter.
+// Stats values are empty; only map keys are used (UniqueTypes / JobsByType) for
+// column filter dropdowns. Pass an ID-scoped filter in history mode so options
+// reflect that task's variants without applying type/job/result narrowing.
+func (s *SQLite) TypeJobKeys(filter *TaskFilter) (TaskStats, error) {
s.mu.Lock()
defer s.mu.Unlock()
if filter == nil {
filter = &TaskFilter{}
}
- filter.Normalize()
-
- whereClause, args := filter.whereForDate(date)
-
- query := `SELECT id, type, job, info, result, meta, created, started, ended
- FROM tasks ` + whereClause
+ whereClause, args := filter.whereForFilter()
+ query := `SELECT DISTINCT type, job FROM tasks ` + whereClause
rows, err := s.db.Query(query, args...)
if err != nil {
- return TaskCounts{}, [24]TaskCounts{}, err
+ return nil, err
}
defer rows.Close()
- var total TaskCounts
- var hourly [24]TaskCounts
+ data := make(TaskStats)
for rows.Next() {
- var t task.Task
- if err := rows.Scan(
- &t.ID, &t.Type, &t.Job, &t.Info, &t.Result, &t.Meta,
- &t.Created, &t.Started, &t.Ended,
- ); err != nil {
+ var typ, job string
+ if err := rows.Scan(&typ, &job); err != nil {
+ continue
+ }
+ key := strings.TrimRight(typ+":"+job, ":")
+ if key == "" {
continue
}
- addTaskHourlyCounts(t, filter, &total, &hourly)
+ data[key] = &Stats{}
}
-
- return total, hourly, rows.Err()
+ return data, rows.Err()
}
// GetTaskRecapByDate creates a recap of tasks for a specific date
@@ -474,12 +469,8 @@ func (s *SQLite) GetTaskRecapByDate(date time.Time) (TaskStats, error) {
continue
}
- job := t.Job
- if job == "" {
- v, _ := url.ParseQuery(t.Meta)
- job = v.Get("job")
- }
- key := strings.TrimRight(t.Type+":"+job, ":")
+ workflow.NormalizeJob(&t)
+ key := strings.TrimRight(t.Type+":"+t.Job, ":")
stat, found := data[key]
if !found {
stat = &Stats{
@@ -497,14 +488,3 @@ func (s *SQLite) GetTaskRecapByDate(date time.Time) (TaskStats, error) {
return TaskStats(data), nil
}
-
-// extractJobFromTask is a helper function to get job from task
-func extractJobFromTask(t task.Task) string {
- job := t.Job
- if job == "" {
- if meta, err := url.ParseQuery(t.Meta); err == nil {
- job = meta.Get("job")
- }
- }
- return job
-}
diff --git a/apps/flowlord/sqlite/tasks_test.go b/apps/flowlord/sqlite/tasks_test.go
index 79a2046..f13b2ca 100644
--- a/apps/flowlord/sqlite/tasks_test.go
+++ b/apps/flowlord/sqlite/tasks_test.go
@@ -9,7 +9,7 @@ import (
"github.com/pcelvng/task/bus/nop"
)
-func TestGetTasksByDate(t *testing.T) {
+func TestGetTasks(t *testing.T) {
db := &SQLite{LocalPath: ":memory:"}
if err := db.initDB(); err != nil {
t.Fatalf("initDB: %v", err)
@@ -27,9 +27,9 @@ func TestGetTasksByDate(t *testing.T) {
db.Add(task.Task{ID: "shared", Type: "zebra", Job: "load", Created: "2024-01-15T14:00:00Z", Result: task.ErrResult})
t.Run("sort type asc", func(t *testing.T) {
- tasks, _, err := db.GetTasksByDate(day, &TaskFilter{Sort: "type", Direction: "asc", Page: 1, Limit: 10})
+ tasks, _, err := db.GetTasks(&TaskFilter{Date: &day, Sort: "type", Direction: "asc", Page: 1, Limit: 10})
if err != nil {
- t.Fatalf("GetTasksByDate: %v", err)
+ t.Fatalf("GetTasks: %v", err)
}
if len(tasks) < 3 {
t.Fatalf("expected at least 3 tasks, got %d", len(tasks))
@@ -45,9 +45,9 @@ func TestGetTasksByDate(t *testing.T) {
})
t.Run("default created desc", func(t *testing.T) {
- tasks, _, err := db.GetTasksByDate(day, &TaskFilter{Page: 1, Limit: 10})
+ tasks, _, err := db.GetTasks(&TaskFilter{Date: &day, Page: 1, Limit: 10})
if err != nil {
- t.Fatalf("GetTasksByDate default: %v", err)
+ t.Fatalf("GetTasks default: %v", err)
}
if len(tasks) == 0 {
t.Fatal("expected tasks")
@@ -58,8 +58,8 @@ func TestGetTasksByDate(t *testing.T) {
})
t.Run("multi type and job", func(t *testing.T) {
- tasks, count, err := db.GetTasksByDate(day, &TaskFilter{
- Type: []string{"alpha", "zebra"}, Job: []string{"load"}, Page: 1, Limit: 10,
+ tasks, count, err := db.GetTasks(&TaskFilter{
+ Date: &day, Type: []string{"alpha", "zebra"}, Job: []string{"load"}, Page: 1, Limit: 10,
})
if err != nil {
t.Fatalf("multi type: %v", err)
@@ -70,8 +70,8 @@ func TestGetTasksByDate(t *testing.T) {
})
t.Run("multi result", func(t *testing.T) {
- _, count, err := db.GetTasksByDate(day, &TaskFilter{
- Result: []string{"error", "complete"}, Type: []string{"alpha"}, Page: 1, Limit: 10,
+ _, count, err := db.GetTasks(&TaskFilter{
+ Date: &day, Result: []string{"error", "complete"}, Type: []string{"alpha"}, Page: 1, Limit: 10,
})
if err != nil {
t.Fatalf("multi result: %v", err)
@@ -82,8 +82,8 @@ func TestGetTasksByDate(t *testing.T) {
})
t.Run("multi id", func(t *testing.T) {
- tasks, count, err := db.GetTasksByDate(day, &TaskFilter{
- ID: []string{"a", "c"}, Page: 1, Limit: 10,
+ tasks, count, err := db.GetTasks(&TaskFilter{
+ Date: &day, ID: []string{"a", "c"}, Page: 1, Limit: 10,
})
if err != nil {
t.Fatalf("multi id: %v", err)
@@ -94,8 +94,8 @@ func TestGetTasksByDate(t *testing.T) {
})
t.Run("id and type", func(t *testing.T) {
- tasks, count, err := db.GetTasksByDate(day, &TaskFilter{
- ID: []string{"shared"}, Type: []string{"zebra"}, Page: 1, Limit: 10,
+ tasks, count, err := db.GetTasks(&TaskFilter{
+ Date: &day, ID: []string{"shared"}, Type: []string{"zebra"}, Page: 1, Limit: 10,
})
if err != nil {
t.Fatalf("id+type: %v", err)
@@ -106,84 +106,103 @@ func TestGetTasksByDate(t *testing.T) {
})
}
-func TestGetHourlyCountsByDate(t *testing.T) {
+func TestGetTasks_idHistory(t *testing.T) {
db := &SQLite{LocalPath: ":memory:"}
if err := db.initDB(); err != nil {
t.Fatalf("initDB: %v", err)
}
defer db.Close()
- day := time.Date(2024, 1, 15, 0, 0, 0, 0, time.UTC)
- db.Add(task.Task{
- ID: "a", Type: "alpha", Job: "load",
- Created: "2024-01-15T10:00:00Z", Meta: "cron=2024-01-15T10",
- Result: task.CompleteResult,
- })
- db.Add(task.Task{
- ID: "b", Type: "alpha", Job: "load",
- Created: "2024-01-15T11:00:00Z", Meta: "cron=2024-01-15T11",
- Result: task.ErrResult,
- })
- db.Add(task.Task{
- ID: "c", Type: "beta", Job: "check",
- Created: "2024-01-15T12:00:00Z", Meta: "cron=2024-01-15T12",
- Result: task.CompleteResult,
- })
+ day15 := time.Date(2024, 1, 15, 0, 0, 0, 0, time.UTC)
+ db.Add(task.Task{ID: "pipe-1", Type: "alpha", Job: "load", Created: "2024-01-15T10:00:00Z", Result: task.ErrResult})
+ db.Add(task.Task{ID: "pipe-1", Type: "alpha", Job: "load", Created: "2024-01-16T09:00:00Z", Result: task.CompleteResult})
+ db.Add(task.Task{ID: "other", Type: "alpha", Job: "load", Created: "2024-01-15T11:00:00Z", Result: task.CompleteResult})
- type input struct {
- filter TaskFilter
- }
- type expect struct {
- total TaskCounts
- hour10 TaskCounts
- hour11 TaskCounts
- hour12 TaskCounts
+ type output struct {
+ Count int
+ IDs []string
}
-
- fn := func(in input) (expect, error) {
- total, hourly, err := db.GetHourlyCountsByDate(day, &in.filter)
+ fn := func(filter TaskFilter) (output, error) {
+ filter.Page = 1
+ filter.Limit = 10
+ tasks, count, err := db.GetTasks(&filter)
if err != nil {
- return expect{}, err
+ return output{}, err
+ }
+ ids := make([]string, len(tasks))
+ for i, tv := range tasks {
+ ids[i] = tv.ID + "@" + tv.Created[:10]
}
- return expect{
- total: total,
- hour10: hourly[10],
- hour11: hourly[11],
- hour12: hourly[12],
- }, nil
+ return output{Count: count, IDs: ids}, nil
}
-
- cases := trial.Cases[input, expect]{
- "id only": {
- Input: input{filter: TaskFilter{ID: []string{"a"}}},
- Expected: expect{
- total: TaskCounts{Total: 1, Completed: 1},
- hour10: TaskCounts{Total: 1, Completed: 1},
+ cases := trial.Cases[TaskFilter, output]{
+ "nil date returns all attempts for id": {
+ Input: TaskFilter{ID: []string{"pipe-1"}},
+ Expected: output{
+ Count: 2,
+ IDs: []string{"pipe-1@2024-01-16", "pipe-1@2024-01-15"},
},
},
- "id and result": {
- Input: input{filter: TaskFilter{
- ID: []string{"a", "b"}, Result: []string{"error"},
- }},
- Expected: expect{
- total: TaskCounts{Total: 1, Error: 1},
- hour11: TaskCounts{Total: 1, Error: 1},
+ "date scope still day-bounded without id": {
+ Input: TaskFilter{Date: &day15},
+ Expected: output{
+ Count: 2,
+ IDs: []string{"other@2024-01-15", "pipe-1@2024-01-15"},
},
},
- "id and type": {
- Input: input{filter: TaskFilter{
- ID: []string{"c"}, Type: []string{"beta"},
- }},
- Expected: expect{
- total: TaskCounts{Total: 1, Completed: 1},
- hour12: TaskCounts{Total: 1, Completed: 1},
+ "date scope with id stays on that day": {
+ Input: TaskFilter{Date: &day15, ID: []string{"pipe-1"}},
+ Expected: output{
+ Count: 1,
+ IDs: []string{"pipe-1@2024-01-15"},
},
},
- "id and type mismatch": {
- Input: input{filter: TaskFilter{
- ID: []string{"c"}, Type: []string{"alpha"},
- }},
- Expected: expect{},
+ }
+ trial.New(fn, cases).SubTest(t)
+}
+
+func TestTypeJobKeys(t *testing.T) {
+ db := &SQLite{LocalPath: ":memory:"}
+ if err := db.initDB(); err != nil {
+ t.Fatalf("initDB: %v", err)
+ }
+ defer db.Close()
+
+ db.Add(task.Task{ID: "pipe-1", Type: "alpha", Job: "load", Created: "2024-01-15T10:00:00Z", Result: task.ErrResult})
+ db.Add(task.Task{ID: "pipe-1", Type: "alpha", Job: "load", Created: "2024-01-16T09:00:00Z", Result: task.CompleteResult})
+ db.Add(task.Task{ID: "pipe-1", Type: "zebra", Job: "check", Created: "2024-01-16T11:00:00Z", Result: task.CompleteResult})
+ db.Add(task.Task{ID: "other", Type: "alpha", Job: "import", Created: "2024-01-15T11:00:00Z", Result: task.CompleteResult})
+
+ type output struct {
+ Types []string
+ Jobs map[string][]string
+ }
+ fn := func(filter TaskFilter) (output, error) {
+ stats, err := db.TypeJobKeys(&filter)
+ if err != nil {
+ return output{}, err
+ }
+ return output{Types: stats.UniqueTypes(), Jobs: stats.JobsByType()}, nil
+ }
+ cases := trial.Cases[TaskFilter, output]{
+ "history mode empty stats bug: id returns distinct type/job keys": {
+ Input: TaskFilter{ID: []string{"pipe-1"}},
+ Expected: output{
+ Types: []string{"alpha", "zebra"},
+ Jobs: map[string][]string{
+ "alpha": {"load"},
+ "zebra": {"check"},
+ },
+ },
+ },
+ "id scope ignores other tasks": {
+ Input: TaskFilter{ID: []string{"other"}},
+ Expected: output{
+ Types: []string{"alpha"},
+ Jobs: map[string][]string{
+ "alpha": {"import"},
+ },
+ },
},
}
trial.New(fn, cases).SubTest(t)
diff --git a/apps/flowlord/sqlite/workflow.go b/apps/flowlord/sqlite/workflow.go
index 636f1ea..1e8dfee 100644
--- a/apps/flowlord/sqlite/workflow.go
+++ b/apps/flowlord/sqlite/workflow.go
@@ -134,12 +134,7 @@ func (s *SQLite) Get(t task.Task) PhaseDB {
s.mu.Lock()
defer s.mu.Unlock()
- values, _ := url.ParseQuery(t.Meta)
- //key := values.Get("workflow")
job := t.Job
- if job == "" {
- job = values.Get("job")
- }
key := t.Type
if job != "" {
key += ":" + job
@@ -187,9 +182,6 @@ func (s *SQLite) Children(t task.Task) []Phase {
values, _ := url.ParseQuery(t.Meta)
key := values.Get("workflow")
job := t.Job
- if job == "" {
- job = values.Get("job")
- }
if key == "" {
return nil
diff --git a/apps/flowlord/sqlite/workflow_test.go b/apps/flowlord/sqlite/workflow_test.go
index 45624df..7432fa5 100644
--- a/apps/flowlord/sqlite/workflow_test.go
+++ b/apps/flowlord/sqlite/workflow_test.go
@@ -230,15 +230,15 @@ func TestGet(t *testing.T) {
Expected: Phase{Task: "task2", DependsOn: "task1"},
},
"task=t2 with job=j1": {
- Input: task.Task{Type: "t2", Meta: "workflow=w2job.toml&job=j1"},
+ Input: task.Task{Type: "t2", Job: "j1", Meta: "workflow=w2job.toml"},
Expected: Phase{Task: "t2:j1", Rule: "job=j1"},
},
"job does not exist": {
- Input: task.Task{Type: "t2", Meta: "workflow=w2job.toml&job=invalid"},
+ Input: task.Task{Type: "t2", Job: "invalid", Meta: "workflow=w2job.toml"},
Expected: Phase{},
},
"wildcard search": {
- Input: task.Task{Type: "t2", Meta: "workflow=*&job=j3"},
+ Input: task.Task{Type: "t2", Job: "j3", Meta: "workflow=*"},
Expected: Phase{Task: "t2:j3"},
},
"wildcard with same task in different files": { // picks first match, results will vary
@@ -290,7 +290,7 @@ func TestChildren(t *testing.T) {
Expected: []Phase{},
},
"task1:j4": {
- Input: task.Task{Type: "task1", Meta: "workflow=workflow.toml&job=j4"},
+ Input: task.Task{Type: "task1", Job: "j4", Meta: "workflow=workflow.toml"},
Expected: []Phase{
{Task: "task2", DependsOn: "task1"},
{Task: "task5", DependsOn: "task1:j4"},
diff --git a/apps/flowlord/taskmaster.go b/apps/flowlord/taskmaster.go
index 2aa1049..5562f66 100644
--- a/apps/flowlord/taskmaster.go
+++ b/apps/flowlord/taskmaster.go
@@ -25,6 +25,7 @@ import (
"github.com/pcelvng/task-tools/file"
"github.com/pcelvng/task-tools/slack"
"github.com/pcelvng/task-tools/tmpl"
+ "github.com/pcelvng/task-tools/workflow"
)
var cronParser = cron.NewParser(cron.SecondOptional | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
@@ -215,7 +216,7 @@ func pName(topic, job string) string {
}
func (tm *taskMaster) getAllChildren(topic, workflow, job string) (s []string) {
- for _, c := range tm.taskCache.Children(task.Task{Type: topic, Meta: "workflow=" + workflow + "&job=" + job}) {
+ for _, c := range tm.taskCache.Children(task.Task{Type: topic, Job: job, Meta: "workflow=" + workflow}) {
job := strings.Trim(c.Topic()+":"+c.Job(), ":")
if children := tm.getAllChildren(c.Task, workflow, c.Job()); len(children) > 0 {
job += " ➞ " + strings.Join(children, " ➞ ")
@@ -395,6 +396,7 @@ func (tm *taskMaster) schedule() (err error) {
// 2. start any downstream tasks
// Send retry failed tasks to tm.failedTopic (only if the phase exists in the workflow)
func (tm *taskMaster) Process(t *task.Task) error {
+ workflow.NormalizeJob(t)
meta, _ := url.ParseQuery(t.Meta)
tm.taskCache.Add(*t)
// attempt to retry
@@ -619,14 +621,8 @@ func (tm *taskMaster) formatImmediateAlert(tsk task.Task) string {
}
key := tsk.Type
- job := tsk.Job
- if job == "" {
- if meta, err := url.ParseQuery(tsk.Meta); err == nil {
- job = meta.Get("job")
- }
- }
- if job != "" {
- key += ":" + job
+ if tsk.Job != "" {
+ key += ":" + tsk.Job
}
message.WriteString(fmt.Sprintf("%s | %s\n", key, tsk.Msg))
return message.String()
diff --git a/apps/flowlord/taskmaster_test.go b/apps/flowlord/taskmaster_test.go
index fc74ee3..02808da 100644
--- a/apps/flowlord/taskmaster_test.go
+++ b/apps/flowlord/taskmaster_test.go
@@ -153,6 +153,7 @@ func TestTaskMaster_Process(t *testing.T) {
"task1:j4 complete": {
Input: task.Task{
Type: "task1",
+ Job: "t2",
Info: "?date=2019-12-12",
Result: task.CompleteResult,
ID: "UUID_task1",
@@ -214,6 +215,7 @@ func TestTaskMaster_Process(t *testing.T) {
"task6 requires file": {
Input: task.Task{
Type: "task5",
+ Job: "t5",
ID: "ID",
Meta: "workflow=f1.toml&file=file.txt&job=t5",
Result: task.CompleteResult,
diff --git a/workflow/workflow.go b/workflow/workflow.go
index 826d3e6..cbc74f7 100644
--- a/workflow/workflow.go
+++ b/workflow/workflow.go
@@ -41,6 +41,21 @@ func (p Phase) Job() string {
return ""
}
+// NormalizeJob copies job from Meta into Job when Job is empty.
+// Prefer setting Job at task creation; this covers legacy inbound tasks.
+func NormalizeJob(t *task.Task) {
+ if t == nil || t.Job != "" || t.Meta == "" {
+ return
+ }
+ values, err := url.ParseQuery(t.Meta)
+ if err != nil {
+ return
+ }
+ if j := values.Get("job"); j != "" {
+ t.Job = j
+ }
+}
+
// Topic portion of the Task
func (p Phase) Topic() string {
s := strings.Split(p.Task, ":")
@@ -117,9 +132,6 @@ func (c *Cache) Get(t task.Task) Phase {
values, _ := url.ParseQuery(t.Meta)
key := values.Get("workflow")
job := t.Job
- if job == "" {
- job = values.Get("job")
- }
if key == "*" { // search all workflows for first match
for _, phases := range c.Workflows {
@@ -166,9 +178,6 @@ func (c *Cache) Children(t task.Task) []Phase {
result := make([]Phase, 0)
key := values.Get("workflow")
- if t.Job == "" {
- t.Job = values.Get("job")
- }
if key == "" {
return nil
}
diff --git a/workflow/workflow_test.go b/workflow/workflow_test.go
index bf67191..b739f63 100644
--- a/workflow/workflow_test.go
+++ b/workflow/workflow_test.go
@@ -204,15 +204,15 @@ func TestGet(t *testing.T) {
Expected: Phase{Task: "task2", DependsOn: "task1"},
},
"task=t2 with job=j1": {
- Input: task.Task{Type: "t2", Meta: "workflow=w2job.toml&job=j1"},
+ Input: task.Task{Type: "t2", Job: "j1", Meta: "workflow=w2job.toml"},
Expected: Phase{Task: "t2", Rule: "job=j1"},
},
"job does not exist": {
- Input: task.Task{Type: "t2", Meta: "workflow=w2job.toml&job=invalid"},
+ Input: task.Task{Type: "t2", Job: "invalid", Meta: "workflow=w2job.toml"},
Expected: Phase{},
},
"wildcard search": {
- Input: task.Task{Type: "t2", Meta: "workflow=*&job=j3"},
+ Input: task.Task{Type: "t2", Job: "j3", Meta: "workflow=*"},
Expected: Phase{Task: "t2:j3"},
},
"wildcard with same task in different files": { // picks first match, results will vary
@@ -264,7 +264,7 @@ func TestChildren(t *testing.T) {
Expected: []Phase{},
},
"task1:j4": {
- Input: task.Task{Type: "task1", Meta: "workflow=workflow.toml&job=j4"},
+ Input: task.Task{Type: "task1", Job: "j4", Meta: "workflow=workflow.toml"},
Expected: []Phase{
{Task: "task2", DependsOn: "task1"},
{Task: "task5", DependsOn: "task1:j4"},
@@ -309,3 +309,25 @@ func TestCache_FilePath(t *testing.T) {
}
trial.New(fn, cases).SubTest(t)
}
+
+func TestNormalizeJob(t *testing.T) {
+ fn := func(in task.Task) (string, error) {
+ NormalizeJob(&in)
+ return in.Job, nil
+ }
+ cases := trial.Cases[task.Task, string]{
+ "keeps existing Job": {
+ Input: task.Task{Job: "keep", Meta: "job=meta"},
+ Expected: "keep",
+ },
+ "copies job from Meta": {
+ Input: task.Task{Meta: "workflow=w.toml&job=from-meta"},
+ Expected: "from-meta",
+ },
+ "empty when neither set": {
+ Input: task.Task{Meta: "workflow=w.toml"},
+ Expected: "",
+ },
+ }
+ trial.New(fn, cases).SubTest(t)
+}