18144468504

Committed 30 Sep 2025 09:52PM UTC coverage: 59.573% (-0.9%) from 60.451%

Build # 18144468504

Build Type

push

github

Committed by

YoSev

Commit Message

test: fix tests after label rebase

Run Details

12 of 13 new or added lines in 7 files covered. (92.31%)

282 existing lines in 15 files now uncovered.

2346 of 3938 relevant lines covered (59.57%)

15.25 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

68.25

/internal/database/repository/task.go

package repository

import (
        "errors"
        "time"

        "github.com/welovemedia/ffmate/v2/internal/database/model"
        "github.com/welovemedia/ffmate/v2/internal/dto"
        "gorm.io/gorm"
        "goyave.dev/goyave/v5/database"
)

type Task struct {
        DB *gorm.DB
}

func (r *Task) Setup() *Task {
        _ = r.DB.AutoMigrate(&model.Task{})
        return r
}

func (r *Task) First(uuid string) (*model.Task, error) {
        var task model.Task
        result := r.DB.Preload("Client").Preload("Labels").Where("uuid = ?", uuid).First(&task)
        if result.Error != nil {
                if errors.Is(result.Error, gorm.ErrRecordNotFound) {
                        return nil, nil
                }
                return nil, result.Error
        }
        return &task, nil
}

func (r *Task) Delete(w *model.Task) error {
        r.DB.Delete(w)
        return r.DB.Error
}

func (r *Task) List(page int, perPage int) (*[]model.Task, int64, error) {
        var tasks = &[]model.Task{}
        tx := r.DB.Preload("Client").Preload("Labels").Order("created_at DESC")
        d := database.NewPaginator(tx, page+1, perPage, tasks)
        err := d.Find()
        return d.Records, d.Total, err
}

func (r *Task) ListByBatch(uuid string, page int, perPage int) (*[]model.Task, int64, error) {
        var tasks = &[]model.Task{}
        tx := r.DB.Preload("Client").Preload("Labels").Order("created_at DESC").Where("batch = ?", uuid)
        d := database.NewPaginator(tx, page+1, perPage, tasks)
        err := d.Find()
        return d.Records, d.Total, err
}

func (r *Task) Add(newTask *model.Task) (*model.Task, error) {
        db := r.DB.Preload("Labels").Create(newTask)

        for i := range newTask.Labels {
                _ = r.DB.FirstOrCreate(&newTask.Labels[i], model.Label{Value: newTask.Labels[i].Value})
        }

        _ = r.DB.Model(newTask).Association("Labels").Replace(newTask.Labels)

        if db.Error != nil {
                return newTask, db.Error
        }
        return r.First(newTask.UUID)
}

func (r *Task) Update(task *model.Task) (*model.Task, error) {
        task.Client = nil // will be re-linked during save
        db := r.DB.Session(&gorm.Session{FullSaveAssociations: true}).Preload("Labels").Save(task)
        if db.Error != nil {
                return task, db.Error
        }
        return r.First(task.UUID)
}

func (r *Task) Count() (int64, error) {
        var count int64
        db := r.DB.Model(&model.Task{}).Count(&count)
        return count, db.Error
}

func (r *Task) FailRunningTasksForStartingClient(identifier string) ([]model.Task, error) {
        var tasks []model.Task
        err := r.DB.Raw(`
                UPDATE tasks
                SET status = ?, error = ?, finished_at = ?,remaining = -1, progress = 100
                WHERE status = ? AND client_identifier = ?
                RETURNING *;
        `, dto.DoneError, "client disconnected during execution", time.Now().UnixMilli(), dto.Running, identifier).Scan(&tasks).Error

        return tasks, err
}

func (r *Task) FailRunningTasksForOfflineClients() ([]model.Task, error) {
        threshold := time.Now().Add(-60 * time.Second).UnixMilli() // int64

        now := time.Now().UnixMilli()
        var tasks []model.Task

        err := r.DB.Raw(`
                UPDATE tasks
                SET status = ?, error = ?, finished_at = ?, remaining = -1, progress = 100
                WHERE status = ?
                  AND client_identifier IN (
                      SELECT identifier
                      FROM client
                      WHERE last_seen < ?
                  )
                RETURNING *;
        `, dto.DoneError, "client disconnected during execution", now, dto.Running, threshold).
                Scan(&tasks).Error

        return tasks, err
}

func (r *Task) CountUnfinishedByBatch(uuid string) (int64, error) {
        var count int64
        db := r.DB.Model(&model.Task{}).Where("batch = ? and status != 'DONE_SUCCESSFUL' and status != 'DONE_ERROR' and status != 'DONE_CANCELED'", uuid).Count(&count)
        return count, db.Error
}

/**
 * Stats (systray) related methods
 */

type statusCount struct {
        Status string
        Count  int
}

func (r *Task) CountAllStatus() (queued, running, doneSuccessful, doneError, doneCanceled int, err error) {
        var counts []statusCount

        r.DB.Model(&model.Task{}).
                Select("status, COUNT(*) as count").
                Group("status").
                Find(&counts)

        err = r.DB.Error

        for _, r := range counts {
                switch r.Status {
                case "QUEUED":
                        queued = r.Count
                case "RUNNING", "PRE_PROCESSING", "POST_PROCESSING":
                        running = r.Count
                case "DONE_SUCCESSFUL":
                        doneSuccessful = r.Count
                case "DONE_ERROR":
                        doneError = r.Count
                case "DONE_CANCELED":
                        doneCanceled = r.Count
                }
        }

        return
}

/**
 * Processing related methods
 */

func (r *Task) NextQueued(amount int, clientLabels dto.Labels) (*[]model.Task, error) {
        var tasks []model.Task

        err := r.DB.Transaction(func(tx *gorm.DB) error {
                // Subquery: task IDs with at least one overlapping label
                sub := tx.Model(&model.Task{}).
                        Select("DISTINCT tasks.id").
                        Joins("JOIN task_labels tl ON tl.task_id = tasks.id").
                        Joins("JOIN labels l ON l.id = tl.label_id").
                        Where("tasks.status = ?", dto.Queued).
                        Where("l.value IN ?", clientLabels).
                        Order("tasks.priority DESC, tasks.created_at ASC").
                        Limit(amount)

                // Load tasks
                if err := tx.Preload("Labels").Where("tasks.id IN (?)", sub).Find(&tasks).Error; err != nil {
                        return err
                }

                if len(tasks) == 0 {
                        return gorm.ErrRecordNotFound
                }

                // Update selected tasks to RUNNING
                ids := make([]uint, len(tasks))
                for i, t := range tasks {
                        ids[i] = t.ID
                }

                if err := tx.Model(&model.Task{}).
                        Where("id IN ?", ids).
                        Update("status", dto.Running).Error; err != nil {
                        return err
                }

                return nil
        })

        if errors.Is(err, gorm.ErrRecordNotFound) {
                return nil, nil
        }

        return &tasks, err
}

/**
 * Stats (telemetry) related methods
 */

func (r *Task) CountAllBySource(source string) (int64, error) {
        var count int64
        db := r.DB.Model(&model.Task{}).Where("source = ?", source).Count(&count)
        return count, db.Error
}

func (r *Task) CountByStatus(status dto.TaskStatus) (int64, error) {
        var count int64
        db := r.DB.Model(&model.Task{}).Where("status = ?", status).Count(&count)
        return count, db.Error
}
func (r *Task) CountDeletedByStatus(status dto.TaskStatus) (int64, error) {
        var count int64
        db := r.DB.Unscoped().Model(&model.Task{}).Unscoped().Where("status = ? AND deleted_at IS NOT NULL", status).Count(&count)
        return count, db.Error
}

func (r *Task) CountDeleted() (int64, error) {
        var count int64
        db := r.DB.Unscoped().Model(&model.Task{}).Unscoped().Where("deleted_at IS NOT NULL").Count(&count)
        return count, db.Error
}

1	package repository
2
3	import (
4	"errors"
5	"time"
6
7	"github.com/welovemedia/ffmate/v2/internal/database/model"
8	"github.com/welovemedia/ffmate/v2/internal/dto"
9	"gorm.io/gorm"
10	"goyave.dev/goyave/v5/database"
11	)
12
13	type Task struct {
14	DB *gorm.DB
15	}
16
17	func (r Task) Setup() Task {	49✔
18	_ = r.DB.AutoMigrate(&model.Task{})	49✔
19	return r	49✔
20	}	49✔
21
22	func (r Task) First(uuid string) (model.Task, error) {	28✔
23	var task model.Task	28✔
24	result := r.DB.Preload("Client").Preload("Labels").Where("uuid = ?", uuid).First(&task)	28✔
25	if result.Error != nil {	29✔
26	if errors.Is(result.Error, gorm.ErrRecordNotFound) {	2✔
27	return nil, nil	1✔
28	}	1✔
UNCOV 29	return nil, result.Error	×
30	}
31	return &task, nil	27✔
32	}
33
34	func (r Task) Delete(w model.Task) error {	1✔
35	r.DB.Delete(w)	1✔
36	return r.DB.Error	1✔
37	}	1✔
38
39	func (r Task) List(page int, perPage int) ([]model.Task, int64, error) {	2✔
40	var tasks = &[]model.Task{}	2✔
41	tx := r.DB.Preload("Client").Preload("Labels").Order("created_at DESC")	2✔
42	d := database.NewPaginator(tx, page+1, perPage, tasks)	2✔
43	err := d.Find()	2✔
44	return d.Records, d.Total, err	2✔
45	}	2✔
46
47	func (r Task) ListByBatch(uuid string, page int, perPage int) ([]model.Task, int64, error) {	1✔
48	var tasks = &[]model.Task{}	1✔
49	tx := r.DB.Preload("Client").Preload("Labels").Order("created_at DESC").Where("batch = ?", uuid)	1✔
50	d := database.NewPaginator(tx, page+1, perPage, tasks)	1✔
51	err := d.Find()	1✔
52	return d.Records, d.Total, err	1✔
53	}	1✔
54
55	func (r Task) Add(newTask model.Task) (*model.Task, error) {	21✔
56	db := r.DB.Preload("Labels").Create(newTask)	21✔
57		21✔
58	for i := range newTask.Labels {	64✔
59	_ = r.DB.FirstOrCreate(&newTask.Labels[i], model.Label{Value: newTask.Labels[i].Value})	43✔
60	}	43✔
61
62	_ = r.DB.Model(newTask).Association("Labels").Replace(newTask.Labels)	21✔
63		21✔
64	if db.Error != nil {	21✔
UNCOV 65	return newTask, db.Error	×
UNCOV 66	}	×
67	return r.First(newTask.UUID)	21✔
68	}	×
69		×
70	func (r Task) Update(task model.Task) (*model.Task, error) {	2✔
71	task.Client = nil // will be re-linked during save	2✔
72	db := r.DB.Session(&gorm.Session{FullSaveAssociations: true}).Preload("Labels").Save(task)	2✔
73	if db.Error != nil {	2✔
UNCOV 74	return task, db.Error	×
UNCOV 75	}	×
76	return r.First(task.UUID)	2✔
77	}
78		×
79	func (r *Task) Count() (int64, error) {	1✔
80	var count int64	1✔
81	db := r.DB.Model(&model.Task{}).Count(&count)	1✔
82	return count, db.Error	1✔
83	}	1✔
84
85	func (r *Task) FailRunningTasksForStartingClient(identifier string) ([]model.Task, error) {	×
86	var tasks []model.Task	×
87	err := r.DB.Raw(`	×
88	UPDATE tasks	×
89	SET status = ?, error = ?, finished_at = ?,remaining = -1, progress = 100	×
UNCOV 90	WHERE status = ? AND client_identifier = ?	×
91	RETURNING *;	×
92	`, dto.DoneError, "client disconnected during execution", time.Now().UnixMilli(), dto.Running, identifier).Scan(&tasks).Error	×
93		×
94	return tasks, err	×
95	}	×
96		×
97	func (r *Task) FailRunningTasksForOfflineClients() ([]model.Task, error) {	×
98	threshold := time.Now().Add(-60 * time.Second).UnixMilli() // int64	×
99		×
100	now := time.Now().UnixMilli()	×
101	var tasks []model.Task	×
102		×
103	err := r.DB.Raw(`	×
104	UPDATE tasks	×
105	SET status = ?, error = ?, finished_at = ?, remaining = -1, progress = 100	×
106	WHERE status = ?	×
107	AND client_identifier IN (	×
108	SELECT identifier	×
109	FROM client	×
110	WHERE last_seen < ?	×
111	)	×
112	RETURNING *;	×
UNCOV 113	`, dto.DoneError, "client disconnected during execution", now, dto.Running, threshold).	×
UNCOV 114	Scan(&tasks).Error	×
UNCOV 115		×
UNCOV 116	return tasks, err	×
UNCOV 117	}	×
118		×
119	func (r *Task) CountUnfinishedByBatch(uuid string) (int64, error) {	1✔
120	var count int64	1✔
121	db := r.DB.Model(&model.Task{}).Where("batch = ? and status != 'DONE_SUCCESSFUL' and status != 'DONE_ERROR' and status != 'DONE_CANCELED'", uuid).Count(&count)	1✔
122	return count, db.Error	1✔
123	}	1✔
124
125	/**	×
126	* Stats (systray) related methods	×
127	*/	×
UNCOV 128		×
UNCOV 129	type statusCount struct {	×
UNCOV 130	Status string	×
UNCOV 131	Count int	×
UNCOV 132	}	×
UNCOV 133		×
134	func (r *Task) CountAllStatus() (queued, running, doneSuccessful, doneError, doneCanceled int, err error) {	1✔
135	var counts []statusCount	1✔
136		1✔
137	r.DB.Model(&model.Task{}).	1✔
138	Select("status, COUNT(*) as count").	1✔
139	Group("status").	1✔
140	Find(&counts)	1✔
141		1✔
142	err = r.DB.Error	1✔
143		1✔
144	for _, r := range counts {	6✔
145	switch r.Status {	5✔
146	case "QUEUED":	1✔
147	queued = r.Count	1✔
148	case "RUNNING", "PRE_PROCESSING", "POST_PROCESSING":	1✔
149	running = r.Count	1✔
150	case "DONE_SUCCESSFUL":	1✔
151	doneSuccessful = r.Count	1✔
152	case "DONE_ERROR":	1✔
153	doneError = r.Count	1✔
154	case "DONE_CANCELED":	1✔
155	doneCanceled = r.Count	1✔
156	}
157	}
158		×
159	return	1✔
UNCOV 160	}	×
UNCOV 161		×
162	/**
163	* Processing related methods
164	*/
165
166	func (r Task) NextQueued(amount int, clientLabels dto.Labels) ([]model.Task, error) {	4✔
167	var tasks []model.Task	4✔
168		4✔
169	err := r.DB.Transaction(func(tx *gorm.DB) error {	8✔
170	// Subquery: task IDs with at least one overlapping label	4✔
171	sub := tx.Model(&model.Task{}).	4✔
172	Select("DISTINCT tasks.id").	4✔
173	Joins("JOIN task_labels tl ON tl.task_id = tasks.id").	4✔
174	Joins("JOIN labels l ON l.id = tl.label_id").	4✔
175	Where("tasks.status = ?", dto.Queued).	4✔
176	Where("l.value IN ?", clientLabels).	4✔
177	Order("tasks.priority DESC, tasks.created_at ASC").	4✔
178	Limit(amount)	4✔
179		4✔
180	// Load tasks	4✔
181	if err := tx.Preload("Labels").Where("tasks.id IN (?)", sub).Find(&tasks).Error; err != nil {	4✔
182	return err	×
183	}	×
184
185	if len(tasks) == 0 {	6✔
186	return gorm.ErrRecordNotFound	2✔
187	}	2✔
188		×
189	// Update selected tasks to RUNNING	×
190	ids := make([]uint, len(tasks))	2✔
191	for i, t := range tasks {	6✔
192	ids[i] = t.ID	4✔
193	}	4✔
194
195	if err := tx.Model(&model.Task{}).	2✔
196	Where("id IN ?", ids).	2✔
197	Update("status", dto.Running).Error; err != nil {	2✔
UNCOV 198	return err	×
UNCOV 199	}	×
200
201	return nil	2✔
202	})
203
204	if errors.Is(err, gorm.ErrRecordNotFound) {	6✔
205	return nil, nil	2✔
206	}	2✔
207
208	return &tasks, err	2✔
209	}
210
211	/**
212	* Stats (telemetry) related methods
213	*/
214
215	func (r *Task) CountAllBySource(source string) (int64, error) {	2✔
216	var count int64	2✔
217	db := r.DB.Model(&model.Task{}).Where("source = ?", source).Count(&count)	2✔
218	return count, db.Error	2✔
219	}	2✔
220
221	func (r *Task) CountByStatus(status dto.TaskStatus) (int64, error) {	5✔
222	var count int64	5✔
223	db := r.DB.Model(&model.Task{}).Where("status = ?", status).Count(&count)	5✔
224	return count, db.Error	5✔
225	}	5✔
226	func (r *Task) CountDeletedByStatus(status dto.TaskStatus) (int64, error) {	3✔
227	var count int64	3✔
228	db := r.DB.Unscoped().Model(&model.Task{}).Unscoped().Where("status = ? AND deleted_at IS NOT NULL", status).Count(&count)	3✔
229	return count, db.Error	3✔
230	}	3✔
231
232	func (r *Task) CountDeleted() (int64, error) {	1✔
233	var count int64	1✔
234	db := r.DB.Unscoped().Model(&model.Task{}).Unscoped().Where("deleted_at IS NOT NULL").Count(&count)	1✔
235	return count, db.Error	1✔
236	}	1✔

welovemedia / ffmate / 18144468504

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous