70 lines
1.7 KiB
Go
70 lines
1.7 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
)
|
|
|
|
type Work struct {
|
|
AbandonedMerges []string
|
|
StartMerges []ClaimedMerge
|
|
FinalizeMerges []FinalizingMerge
|
|
StartDiscards []string
|
|
FinishDiscards []string
|
|
}
|
|
|
|
func (s *Store) Coordinate(ctx context.Context) (Work, error) {
|
|
work := Work{}
|
|
err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
|
var err error
|
|
work.AbandonedMerges, err = abandonMerges(ctx, tx, emit)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
err = claimHeldMerges(ctx, tx, emit)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
work.StartMerges, err = startableMerges(ctx, tx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
work.FinalizeMerges, err = finalizingMerges(ctx, tx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.ExecContext(ctx, `UPDATE tasks SET discard_state = 'closing'
|
|
WHERE status = 'active' AND discard_state = 'running'
|
|
AND NOT EXISTS (SELECT 1 FROM agent_sessions s WHERE s.id = tasks.discard_session AND s.state = 'running')`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
work.StartDiscards, err = queryIDs(ctx, tx, `SELECT id FROM tasks WHERE status = 'active' AND discard_state = 'held' AND NOT `+implementerRunning)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
work.FinishDiscards, err = queryIDs(ctx, tx, `SELECT id FROM tasks WHERE status = 'active' AND discard_state = 'closing'`)
|
|
return err
|
|
})
|
|
return work, err
|
|
}
|
|
|
|
func queryIDs(ctx context.Context, tx *sql.Tx, query string) ([]string, error) {
|
|
rows, err := tx.QueryContext(ctx, query)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
ids := []string{}
|
|
for rows.Next() {
|
|
id := ""
|
|
err = rows.Scan(&id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ids = append(ids, id)
|
|
}
|
|
return ids, rows.Err()
|
|
}
|