Janady13/missionmatch / internal/modules/recruiting/task_service.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / internal/modules/recruiting/task_service.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	internal/modules/recruiting/task_service.go
Original SHA-256	d2ddb9f55c8d87e332a5bcd0a47fa770734a7388b5c3bfc1ae4d34bd293e7aff
Git blob SHA	6283f2b668b1fa417c32925b4c87bfa1adb5179f
Reading-copy SHA-256	d2ddb9f55c8d87e332a5bcd0a47fa770734a7388b5c3bfc1ae4d34bd293e7aff
Original source ↗
Every recorded occurrence

These 2 records contain identical original source bytes. Their paths and revisions remain part of the archive.
Original record	Branch or archive	Revision / identifier	Original location
Janady13/missionmatch / internal/modules/recruiting/task_service.go	feature/opportunity-to-outcome	720d47fae476b0d38477ae4401697d372b1ea2bc	Open original
Janady13/missionmatch / internal/modules/recruiting/task_service.go	feature/aioe-upgrade	2178e7298d0e5d9af83afa12402fd587f0ba6ed8	Open original
Recovered source transcript

Download complete source text ↓484 lines · 15.8 KB
package recruiting

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"

	"github.com/missionworkforce/missionmatch/internal/platform/audit"
	"github.com/missionworkforce/missionmatch/internal/platform/security"
	"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)

var taskStatuses = map[string]bool{"open": true, "done": true, "cancelled": true}
var taskPriorities = map[string]bool{"low": true, "normal": true, "high": true}

// Bucket names. "today" is the organization's working day, not a UTC day, which is the distinction that
// makes an evening call count toward the day it was actually made.
const (
	bucketOverdue = "overdue"
	bucketToday   = "today"
	bucketSoon    = "soon"
	bucketLater   = "later"
	bucketNone    = "none"
)

// soonWindowDays is where "soon" ends. Seven days is a working week, which is the horizon a recruiter
// plans over; beyond that the queue is a backlog rather than a plan.
const soonWindowDays = 7

// bucketFor decides which queue a task belongs in.
//
// Every comparison is on whole days in the organization's timezone, never on instants. A task due at
// 9am and one due at 5pm on the same day are both due today, and computing that on instants would put
// the morning one in the past by lunchtime.
//
// dueDays is whole days from today to the due date: negative is overdue, zero is today.
func bucketFor(dueDays *int) string {
	switch {
	case dueDays == nil:
		return bucketNone
	case *dueDays < 0:
		return bucketOverdue
	case *dueDays == 0:
		return bucketToday
	case *dueDays <= soonWindowDays:
		return bucketSoon
	default:
		return bucketLater
	}
}

const taskSelect = `t.id::text, t.title, t.detail, t.status, t.priority,
	t.assigned_to::text, u.full_name, t.due_at, t.completed_at,
	t.candidate_id::text, c.full_name, t.requisition_id::text, r.title,
	t.application_id::text, t.contact_id::text, t.created_at, t.updated_at,
	CASE WHEN t.due_at IS NULL THEN NULL ELSE
	  ((t.due_at AT TIME ZONE rm_org_tz())::date - (now() AT TIME ZONE rm_org_tz())::date)
	END`

const taskFrom = `FROM tasks t
	LEFT JOIN users        u ON u.organization_id = t.organization_id AND u.id = t.assigned_to
	LEFT JOIN candidates   c ON c.organization_id = t.organization_id AND c.id = t.candidate_id
	LEFT JOIN requisitions r ON r.organization_id = t.organization_id AND r.id = t.requisition_id`

// taskOrder is the queue order, and it is total.
//
// Overdue first, then soonest, then undated, then by id. The id tiebreak is what makes paging
// deterministic: without it two tasks sharing a due date can swap between pages, so one is shown twice
// and another never.
const taskOrder = ` ORDER BY (t.status <> 'open'), t.due_at ASC NULLS LAST,
	CASE t.priority WHEN 'high' THEN 0 WHEN 'normal' THEN 1 ELSE 2 END, t.id ASC`

func scanTask(row scannable) (Task, error) {
	var t Task
	var assignee, assigneeName, candID, candName, reqID, reqTitle, appID, contactID *string
	var due, completed *time.Time
	var created, updated time.Time
	var dueDays *int
	if err := row.Scan(&t.ID, &t.Title, &t.Detail, &t.Status, &t.Priority,
		&assignee, &assigneeName, &due, &completed,
		&candID, &candName, &reqID, &reqTitle, &appID, &contactID,
		&created, &updated, &dueDays); err != nil {
		return Task{}, err
	}
	t.AssignedTo, t.AssigneeName = str(assignee), str(assigneeName)
	t.CandidateID, t.CandidateName = str(candID), str(candName)
	t.RequisitionID, t.RequisitionTitle = str(reqID), str(reqTitle)
	t.ApplicationID, t.ContactID = str(appID), str(contactID)
	if due != nil {
		t.DueAt = ts(*due)
	}
	if completed != nil {
		t.CompletedAt = ts(*completed)
	}
	t.CreatedAt, t.UpdatedAt = ts(created), ts(updated)
	t.DueInDays = dueDays
	// A closed task is not in any due bucket. Showing a completed task as overdue is the fastest way to
	// make a queue nobody trusts.
	if t.Status == "open" {
		t.Bucket = bucketFor(dueDays)
	} else {
		t.Bucket = bucketNone
	}
	return t, nil
}

// CreateTask adds work to a queue. Unassigned is permitted, because work often exists before it is
// clear who will do it, and forcing an assignee produces tasks parked on whoever happened to type them.
func (s *Service) CreateTask(ctx context.Context, in CreateTaskInput) (Task, error) {
	if s.db == nil {
		return Task{}, ErrNoDatabase
	}
	title, err := clean("title", in.Title, maxReqTitle)
	if err != nil {
		return Task{}, err
	}
	if title == "" {
		return Task{}, &ValidationError{Field: "title", Message: "is required"}
	}
	detail, err := clean("detail", in.Detail, maxReqDesc)
	if err != nil {
		return Task{}, err
	}
	if in.Priority == "" {
		in.Priority = "normal"
	}
	if !taskPriorities[in.Priority] {
		return Task{}, &ValidationError{Field: "priority", Message: "is not a permitted value"}
	}
	assignee, err := optUUID("assigned_to", in.AssignedTo)
	if err != nil {
		return Task{}, err
	}
	// Default to the person creating it. A task typed with no assignee is almost always for oneself, and
	// leaving it unassigned would bury it in the queue that exists for work nobody owns.
	if assignee == nil {
		assignee = actorPtr(ctx)
	}
	due, err := optTime("due_at", in.DueAt)
	if err != nil {
		return Task{}, err
	}
	candID, err := optUUID("candidate_id", in.CandidateID)
	if err != nil {
		return Task{}, err
	}
	contactID, err := optUUID("contact_id", in.ContactID)
	if err != nil {
		return Task{}, err
	}
	reqID, err := optUUID("requisition_id", in.RequisitionID)
	if err != nil {
		return Task{}, err
	}
	appID, err := optUUID("application_id", in.ApplicationID)
	if err != nil {
		return Task{}, err
	}
	actor := actorPtr(ctx)
	if actor == nil {
		return Task{}, tenant.ErrNoTenant
	}

	// Scanned as a uuid.UUID rather than as text and parsed back with uuid.MustParse. See
	// CreateApplication for the reasoning: the only thing that parse could ever do is panic in a request
	// path, and this one runs after the commit.
	var id uuid.UUID
	err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		e := tx.QueryRow(ctx, `
			INSERT INTO tasks (organization_id, assigned_to, title, detail, priority, due_at,
			                   candidate_id, contact_id, requisition_id, application_id, created_by)
			VALUES (NULLIF(current_setting('app.current_org_id', true), '')::uuid,
			        $1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
			RETURNING id`,
			assignee, title, detail, in.Priority, due, candID, contactID, reqID, appID, actor).Scan(&id)
		if e != nil {
			return translateTaskErr(e)
		}
		nv, _ := json.Marshal(map[string]any{"id": id.String(), "title": title, "assigned_to": in.AssignedTo})
		return s.audit.OnTx(ctx, tx, audit.Event{
			Action: "task.created", ActorUserID: actor,
			EntityType: "task", EntityID: id.String(), NewValues: nv,
		})
	})
	if err != nil {
		return Task{}, err
	}
	return s.taskRow(ctx, id)
}

// UpdateTask edits or closes a task. completed_at is not accepted and not written here: the database
// stamps it when the status becomes done and clears it otherwise, so reopening works from any path and
// the constraint can never be violated by a correct status change.
func (s *Service) UpdateTask(ctx context.Context, id uuid.UUID, in UpdateTaskInput) (Task, error) {
	if s.db == nil {
		return Task{}, ErrNoDatabase
	}
	expected, err := parseToken(in.ExpectedUpdatedAt)
	if err != nil {
		return Task{}, err
	}
	if in.Title != nil {
		t, e := clean("title", *in.Title, maxReqTitle)
		if e != nil {
			return Task{}, e
		}
		if t == "" {
			return Task{}, &ValidationError{Field: "title", Message: "is required"}
		}
		in.Title = &t
	}
	if in.Status != nil && !taskStatuses[*in.Status] {
		return Task{}, &ValidationError{Field: "status", Message: "is not a permitted value"}
	}
	if in.Priority != nil && !taskPriorities[*in.Priority] {
		return Task{}, &ValidationError{Field: "priority", Message: "is not a permitted value"}
	}
	var assignee *uuid.UUID
	if in.AssignedTo != nil {
		if assignee, err = optUUID("assigned_to", *in.AssignedTo); err != nil {
			return Task{}, err
		}
	}
	var due *time.Time
	if in.DueAt != nil {
		if due, err = optTime("due_at", *in.DueAt); err != nil {
			return Task{}, err
		}
	}

	err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		tag, e := tx.Exec(ctx, `
			UPDATE tasks SET
			  title       = COALESCE($2, title),
			  detail      = COALESCE($3, detail),
			  status      = COALESCE($4, status),
			  priority    = COALESCE($5, priority),
			  assigned_to = CASE WHEN $6 THEN $7 ELSE assigned_to END,
			  due_at      = CASE WHEN $8 THEN $9 ELSE due_at END
			WHERE id = $1 AND ($10::timestamptz IS NULL OR updated_at = $10)`,
			id, in.Title, in.Detail, in.Status, in.Priority,
			in.AssignedTo != nil, assignee, in.DueAt != nil, due, tokenOrNil(expected, false))
		if e != nil {
			return translateTaskErr(e)
		}
		if tag.RowsAffected() == 0 {
			var exists bool
			if q := tx.QueryRow(ctx, `SELECT true FROM tasks WHERE id = $1`, id).Scan(&exists); q != nil {
				return ErrNotFound
			}
			return ErrConflict
		}
		nv, _ := json.Marshal(in)
		return s.audit.OnTx(ctx, tx, audit.Event{
			Action: "task.updated", ActorUserID: actorPtr(ctx),
			EntityType: "task", EntityID: id.String(), NewValues: nv,
		})
	})
	if err != nil {
		return Task{}, err
	}
	return s.taskRow(ctx, id)
}

func (s *Service) taskRow(ctx context.Context, id uuid.UUID) (Task, error) {
	var out Task
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		t, e := scanTask(tx.QueryRow(ctx, `SELECT `+taskSelect+` `+taskFrom+` WHERE t.id = $1`, id))
		if e != nil {
			if errors.Is(e, pgx.ErrNoRows) {
				return ErrNotFound
			}
			return translatePG(e)
		}
		out = t
		return nil
	})
	return out, err
}

// TaskQueueView is the work screen: every open task for a scope, bucketed by the organization's
// working day, with the counts that go at the top.
//
// The counts are computed from the same rows that are returned rather than by separate aggregate
// queries. Two queries against a moving table can disagree, and a header that contradicts the list
// below it is worse than no header.
func (s *Service) TaskQueueView(ctx context.Context, assignee string, includeClosed bool) (TaskQueue, error) {
	if s.db == nil {
		return TaskQueue{}, ErrNoDatabase
	}
	where := []string{"true"}
	var args []any
	add := func(clause string, v any) {
		args = append(args, v)
		where = append(where, strings.ReplaceAll(clause, "?", "$"+itoa(len(args))))
	}
	switch assignee {
	case "":
	case "me":
		actor := actorPtr(ctx)
		if actor == nil {
			return TaskQueue{}, tenant.ErrNoTenant
		}
		add("t.assigned_to = ?", *actor)
	case "unassigned":
		where = append(where, "t.assigned_to IS NULL")
	default:
		id, err := uuid.Parse(assignee)
		if err != nil {
			return TaskQueue{}, &ValidationError{Field: "assignee",
				Message: "must be me, unassigned, or a user id"}
		}
		add("t.assigned_to = ?", id)
	}
	if !includeClosed {
		where = append(where, "t.status = 'open'")
	}
	cond := strings.Join(where, " AND ")

	out := TaskQueue{Tasks: []Task{}}
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		if e := tx.QueryRow(ctx, `SELECT rm_org_tz()`).Scan(&out.TimeZone); e != nil {
			return translatePG(e)
		}
		rows, e := tx.Query(ctx, `SELECT `+taskSelect+` `+taskFrom+` WHERE `+cond+taskOrder+` LIMIT 500`, args...)
		if e != nil {
			return translatePG(e)
		}
		defer rows.Close()
		for rows.Next() {
			t, e := scanTask(rows)
			if e != nil {
				return translatePG(e)
			}
			switch t.Bucket {
			case bucketOverdue:
				out.Overdue++
			case bucketToday:
				out.Today++
			case bucketSoon:
				out.Soon++
			case bucketLater:
				out.Later++
			case bucketNone:
				if t.Status == "open" {
					out.NoDueDate++
				}
			}
			if t.AssignedTo == "" && t.Status == "open" {
				out.Unassigned++
			}
			out.Tasks = append(out.Tasks, t)
		}
		return rows.Err()
	})
	if err != nil {
		return TaskQueue{}, err
	}
	return out, nil
}

// ListTasks is the paged list for a subject, used by the candidate and job order screens.
func (s *Service) ListTasks(ctx context.Context, candidateID, requisitionID, applicationID string, limit, offset int) (TaskList, error) {
	if s.db == nil {
		return TaskList{}, ErrNoDatabase
	}
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	if offset < 0 {
		offset = 0
	}
	where := []string{"true"}
	var args []any
	for field, raw := range map[string]string{
		"candidate_id": candidateID, "requisition_id": requisitionID, "application_id": applicationID,
	} {
		if raw == "" {
			continue
		}
		id, err := uuid.Parse(raw)
		if err != nil {
			return TaskList{}, &ValidationError{Field: field, Message: "must be an id"}
		}
		args = append(args, id)
		where = append(where, "t."+field+" = $"+itoa(len(args)))
	}
	cond := strings.Join(where, " AND ")

	out := TaskList{Tasks: []Task{}, Limit: limit, Offset: offset}
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		if e := tx.QueryRow(ctx, `SELECT count(*) `+taskFrom+` WHERE `+cond, args...).Scan(&out.Total); e != nil {
			return translatePG(e)
		}
		rows, e := tx.Query(ctx, `SELECT `+taskSelect+` `+taskFrom+` WHERE `+cond+taskOrder+
			` LIMIT $`+itoa(len(args)+1)+` OFFSET $`+itoa(len(args)+2), append(args, limit, offset)...)
		if e != nil {
			return translatePG(e)
		}
		defer rows.Close()
		for rows.Next() {
			t, e := scanTask(rows)
			if e != nil {
				return translatePG(e)
			}
			out.Tasks = append(out.Tasks, t)
		}
		return rows.Err()
	})
	if err != nil {
		return TaskList{}, err
	}
	return out, nil
}

// SetTimeZone changes the organization's working day. It moves what "today" means for every count in
// the product, so it is an owner level setting rather than a per user preference.
func (s *Service) SetTimeZone(ctx context.Context, tz string) (string, error) {
	if s.db == nil {
		return "", ErrNoDatabase
	}
	// Checked here as well as at the route.
	//
	// The HTTP handler already requires org.settings.write, so over the API this is redundant. It is not
	// redundant for any other caller: a CLI, a scheduled job or a future handler reaching this method
	// directly would otherwise face no check at all, because the only gate lived in the routing layer.
	// Reassign guards itself the same way; this method was the outlier, and an integration test calling
	// the service rather than the endpoint is what exposed it.
	p, ok := tenant.UserFromContext(ctx)
	if !ok || !security.Has(security.Role(p.Role), security.CapOrgSettings) {
		return "", &ValidationError{Field: "time_zone",
			Message: "only an owner can change the organization time zone"}
	}
	tz = strings.TrimSpace(tz)
	if tz == "" {
		return "", &ValidationError{Field: "time_zone", Message: "is required"}
	}
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		tag, e := tx.Exec(ctx, `
			UPDATE organizations SET time_zone = $1, updated_at = now()
			 WHERE id = NULLIF(current_setting('app.current_org_id', true), '')::uuid`, tz)
		if e != nil {
			return translateTaskErr(e)
		}
		if tag.RowsAffected() == 0 {
			return ErrNotFound
		}
		return s.audit.OnTx(ctx, tx, audit.Event{
			Action: "org.time_zone.changed", ActorUserID: actorPtr(ctx),
			EntityType: "organization", EntityID: tz,
		})
	})
	if err != nil {
		return "", err
	}
	return tz, nil
}

func translateTaskErr(err error) error {
	var pg *pgconn.PgError
	if !errors.As(err, &pg) {
		return translatePG(err)
	}
	switch {
	case pg.Code == "22023" && strings.Contains(pg.Message, "IANA"):
		return &ValidationError{Field: "time_zone", Message: pg.Message}
	case pg.Code == "22023":
		return &ValidationError{Field: "time_zone", Message: "a time zone is required"}
	case pg.Code == "23514" && strings.Contains(pg.ConstraintName, "tasks_title_check"):
		return &ValidationError{Field: "title", Message: "must be between 1 and 300 characters"}
	case pg.Code == "23514" && strings.Contains(pg.ConstraintName, "tasks_completed_chk"):
		// Unreachable: the trigger maintains completed_at from the status. Reaching it means the trigger
		// was dropped, which is a defect here rather than bad input.
		return err
	}
	return translatePG(err)
}