BCDA-8215: Remove completed job count application logic (#964)

## 🎫 Ticket

https://jira.cms.gov/browse/BCDA-8215

## 🛠 Changes

- Remove all application logic referencing completed job count

## ℹ️ Context

This column, by it's own commented definition, is not a reliable
indicator of the number of actually completed sub-jobs for an export
job. It appears that its only use is in the in-progress job status
header to indicate roughly how far along the job processing is.

Instead of attempting to maintain this field, we should be relying on
the source of truth (the number of completed job keys). To avoid
confusion and to reduce code complexity, we are removing this column.

Before removing the column from the database, we must stop reading it
from the application.

## 🧪 Validation

- Unit testing

bcda/api/requests.go

@@ -258,7 +258,13 @@ func (h *Handler) JobStatus(w http.ResponseWriter, r *http.Request) {
 		logger.Error(job.Status)
 		h.RespWriter.Exception(r.Context(), w, http.StatusInternalServerError, responseutils.JobFailed, responseutils.DetailJobFailed)
 	case models.JobStatusPending, models.JobStatusInProgress:
-		w.Header().Set("X-Progress", job.StatusMessage())
+		completedJobKeyCount := utils.CountUniq(jobKeys, func(jobKey *models.JobKey) int64 {
+			if jobKey.QueJobID == nil {
+				return -1
+			}
+			return *jobKey.QueJobID
+		})
+		w.Header().Set("X-Progress", job.StatusMessage(completedJobKeyCount))
 		w.WriteHeader(http.StatusAccepted)
 		return
 	case models.JobStatusCompleted:

bcda/api/requests_test.go

@@ -757,15 +757,14 @@ func (s *RequestsTestSuite) TestJobStatusErrorHandling() {
 
 				mockSrv.On("GetJobAndKeys", testUtils.CtxMatcher, uint(1)).Return(
 					&models.Job{
-						ID:                1,
-						ACOID:             uuid.NewRandom(),
-						RequestURL:        requestUrl,
-						Status:            tt.status,
-						TransactionTime:   timestp,
-						JobCount:          100,
-						CompletedJobCount: 100,
-						CreatedAt:         timestp,
-						UpdatedAt:         timestp.Add(time.Duration(tt.timestampOffset)),
+						ID:              1,
+						ACOID:           uuid.NewRandom(),
+						RequestURL:      requestUrl,
+						Status:          tt.status,
+						TransactionTime: timestp,
+						JobCount:        100,
+						CreatedAt:       timestp,
+						UpdatedAt:       timestp.Add(time.Duration(tt.timestampOffset)),
 					},
 					[]*models.JobKey{{
 						ID:           1,
@@ -890,15 +889,14 @@ func (s *RequestsTestSuite) TestJobFailedStatus() {
 			timestp := time.Now()
 			mockSrv.On("GetJobAndKeys", testUtils.CtxMatcher, uint(1)).Return(
 				&models.Job{
-					ID:                1,
-					ACOID:             uuid.NewRandom(),
-					RequestURL:        tt.requestUrl,
-					Status:            tt.status,
-					TransactionTime:   timestp,
-					JobCount:          100,
-					CompletedJobCount: 100,
-					CreatedAt:         timestp,
-					UpdatedAt:         timestp,
+					ID:              1,
+					ACOID:           uuid.NewRandom(),
+					RequestURL:      tt.requestUrl,
+					Status:          tt.status,
+					TransactionTime: timestp,
+					JobCount:        100,
+					CreatedAt:       timestp,
+					UpdatedAt:       timestp,
 				},
 				[]*models.JobKey{{
 					ID:           1,

bcda/models/models.go

@@ -2,6 +2,7 @@ package models
 
 import (
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/pborman/uuid"
@@ -24,20 +25,19 @@ var AllJobStatuses []JobStatus = []JobStatus{JobStatusPending, JobStatusInProgre
 
 type JobStatus string
 type Job struct {
-	ID                uint
-	ACOID             uuid.UUID `json:"aco_id"`
-	RequestURL        string    `json:"request_url"` // request_url
-	Status            JobStatus `json:"status"`      // status
-	TransactionTime   time.Time // most recent data load transaction time from BFD
-	JobCount          int
-	CompletedJobCount int
-	CreatedAt         time.Time
-	UpdatedAt         time.Time
+	ID              uint
+	ACOID           uuid.UUID `json:"aco_id"`
+	RequestURL      string    `json:"request_url"` // request_url
+	Status          JobStatus `json:"status"`      // status
+	TransactionTime time.Time // most recent data load transaction time from BFD
+	JobCount        int
+	CreatedAt       time.Time
+	UpdatedAt       time.Time
 }
 
-func (j *Job) StatusMessage() string {
+func (j *Job) StatusMessage(numCompletedJobKeys int) string {
 	if j.Status == JobStatusInProgress && j.JobCount > 0 {
-		pct := float64(j.CompletedJobCount) / float64(j.JobCount) * 100
+		pct := float64(numCompletedJobKeys) / float64(j.JobCount) * 100
 		return fmt.Sprintf("%s (%d%%)", j.Status, int(pct))
 	}
 
@@ -57,6 +57,10 @@ type JobKey struct {
 	ResourceType string
 }
 
+func (j *JobKey) IsError() bool {
+	return strings.Contains(j.FileName, "-error.ndjson")
+}
+
 // ACO represents an Accountable Care Organization.
 type ACO struct {
 	ID                 uint

bcda/models/models_test.go

@@ -20,14 +20,14 @@ func TestModelsTestSuite(t *testing.T) {
 }
 
 func (s *ModelsTestSuite) TestJobStatusMessage() {
-	j := Job{Status: constants.InProgress, JobCount: 25, CompletedJobCount: 6}
-	assert.Equal(s.T(), "In Progress (24%)", j.StatusMessage())
+	j := Job{Status: constants.InProgress, JobCount: 25}
+	assert.Equal(s.T(), "In Progress (24%)", j.StatusMessage(6))
 
-	j = Job{Status: constants.InProgress, JobCount: 0, CompletedJobCount: 0}
-	assert.Equal(s.T(), constants.InProgress, j.StatusMessage())
+	j = Job{Status: constants.InProgress, JobCount: 0}
+	assert.Equal(s.T(), constants.InProgress, j.StatusMessage(0))
 
-	j = Job{Status: JobStatusCompleted, JobCount: 25, CompletedJobCount: 25}
-	assert.Equal(s.T(), string(JobStatusCompleted), j.StatusMessage())
+	j = Job{Status: JobStatusCompleted, JobCount: 25}
+	assert.Equal(s.T(), string(JobStatusCompleted), j.StatusMessage(25))
 }
 
 func (s *ModelsTestSuite) TestACOBlacklist() {

bcda/models/postgres/repository.go

@@ -367,7 +367,7 @@ func (r *Repository) UpdateSuppressionFileImportStatus(ctx context.Context, file
 	return nil
 }
 
-var jobColumns []string = []string{"id", "aco_id", "request_url", "status", "transaction_time", "job_count", "completed_job_count", "created_at", "updated_at"}
+var jobColumns []string = []string{"id", "aco_id", "request_url", "status", "transaction_time", "job_count", "created_at", "updated_at"}
 
 func (r *Repository) GetJobs(ctx context.Context, acoID uuid.UUID, statuses ...models.JobStatus) ([]*models.Job, error) {
 	s := make([]interface{}, len(statuses))
@@ -425,7 +425,7 @@ func (r *Repository) GetJobByID(ctx context.Context, jobID uint) (*models.Job, e
 	)
 
 	err := r.QueryRowContext(ctx, query, args...).Scan(&j.ID, &j.ACOID, &j.RequestURL, &j.Status, &transactionTime,
-		&j.JobCount, &j.CompletedJobCount, &createdAt, &updatedAt)
+		&j.JobCount, &createdAt, &updatedAt)
 	j.TransactionTime, j.CreatedAt, j.UpdatedAt = transactionTime.Time, createdAt.Time, updatedAt.Time
 
 	if err != nil {
@@ -439,10 +439,10 @@ func (r *Repository) CreateJob(ctx context.Context, j models.Job) (uint, error)
 	// User raw builder since we need to retrieve the associated ID
 	ib := sqlFlavor.NewInsertBuilder().InsertInto("jobs")
 	ib.Cols("aco_id", "request_url", "status",
-		"transaction_time", "job_count", "completed_job_count",
+		"transaction_time", "job_count",
 		"created_at", "updated_at").
 		Values(j.ACOID, j.RequestURL, j.Status,
-			j.TransactionTime, j.JobCount, j.CompletedJobCount,
+			j.TransactionTime, j.JobCount,
 			sqlbuilder.Raw("NOW()"), sqlbuilder.Raw("NOW()"))
 
 	query, args := ib.Build()
@@ -465,7 +465,6 @@ func (r *Repository) UpdateJob(ctx context.Context, j models.Job) error {
 		ub.Assign("status", j.Status),
 		ub.Assign("transaction_time", j.TransactionTime),
 		ub.Assign("job_count", j.JobCount),
-		ub.Assign("completed_job_count", j.CompletedJobCount),
 		ub.Assign("updated_at", sqlbuilder.Raw("NOW()")),
 	)
 	ub.Where(ub.Equal("id", j.ID))
@@ -545,7 +544,7 @@ func (r *Repository) getJobs(ctx context.Context, query string, args ...interfac
 	for rows.Next() {
 		var j models.Job
 		if err = rows.Scan(&j.ID, &j.ACOID, &j.RequestURL, &j.Status, &transactionTime,
-			&j.JobCount, &j.CompletedJobCount, &createdAt, &updatedAt); err != nil {
+			&j.JobCount, &createdAt, &updatedAt); err != nil {
 			return nil, err
 		}
 		j.TransactionTime, j.CreatedAt, j.UpdatedAt = transactionTime.Time, createdAt.Time, updatedAt.Time

bcda/models/postgres/repository_test.go

@@ -683,9 +683,9 @@ func (r *RepositoryTestSuite) TestJobsMethods() {
 
 	defer postgrestest.DeleteACO(r.T(), r.db, aco.UUID)
 
-	failed := models.Job{ACOID: aco.UUID, RequestURL: reqURL, Status: models.JobStatusFailed, JobCount: 10, CompletedJobCount: 20}
-	pending := models.Job{ACOID: aco.UUID, RequestURL: reqURL, Status: models.JobStatusPending, JobCount: 30, CompletedJobCount: 40}
-	completed := models.Job{ACOID: aco.UUID, RequestURL: reqURL, Status: models.JobStatusCompleted, JobCount: 40, CompletedJobCount: 60}
+	failed := models.Job{ACOID: aco.UUID, RequestURL: reqURL, Status: models.JobStatusFailed, JobCount: 10}
+	pending := models.Job{ACOID: aco.UUID, RequestURL: reqURL, Status: models.JobStatusPending, JobCount: 30}
+	completed := models.Job{ACOID: aco.UUID, RequestURL: reqURL, Status: models.JobStatusCompleted, JobCount: 40}
 
 	failed.ID, err = r.repository.CreateJob(ctx, failed)
 	assert.NoError(err)
@@ -755,7 +755,6 @@ func (r *RepositoryTestSuite) TestJobsMethods() {
 	// Account for time precision in postgres
 	failed.TransactionTime = time.Now().Round(time.Millisecond)
 	failed.JobCount = failed.JobCount + 1
-	failed.CompletedJobCount = failed.CompletedJobCount + 1
 	failed.Status = models.JobStatusArchived
 	assert.NoError(r.repository.UpdateJob(ctx, failed))
 
@@ -765,7 +764,6 @@ func (r *RepositoryTestSuite) TestJobsMethods() {
 	assert.Equal(failed.TransactionTime.UTC(), newFailed.TransactionTime.UTC())
 	assert.Equal(failed.Status, newFailed.Status)
 	assert.Equal(failed.JobCount, newFailed.JobCount)
-	assert.Equal(failed.CompletedJobCount, newFailed.CompletedJobCount)
 
 	// Verify that we did not modify other job
 	newCompleted, err := r.repository.GetJobByID(ctx, completed.ID)

bcda/utils/common.go

@@ -115,3 +115,19 @@ func Dedup(slice []string) []string {
 
 	return newSlice
 }
+
+// Count the number of unique values in the slice based on given function
+func CountUniq[S []E, E any, F comparable](arr S, f func(E) F) int {
+	var n int
+	var dupcheck = make(map[F]bool, n)
+
+	for _, val := range arr {
+		comparableVal := f(val)
+		if !dupcheck[comparableVal] {
+			dupcheck[comparableVal] = true
+			n++
+		}
+	}
+
+	return n
+}

bcda/utils/common_test.go

@@ -1,9 +1,10 @@
 package utils
 
 import (
+	"testing"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/suite"
-	"testing"
 )
 
 type CommonTestSuite struct {
@@ -26,6 +27,13 @@ func (s *CommonTestSuite) TestDedup() {
 	assert.Len(s.T(), result, 3)
 }
 
+func (s *CommonTestSuite) TestCountUniq() {
+	firstLetter := func(s string) string { return string(s[0]) }
+	assert.Equal(s.T(), 0, CountUniq([]string{}, firstLetter))
+	assert.Equal(s.T(), 1, CountUniq([]string{"abc", "ab"}, firstLetter))
+	assert.Equal(s.T(), 2, CountUniq([]string{"abc", "bcd", "ab"}, firstLetter))
+}
+
 func TestCommonTestSuite(t *testing.T) {
 	suite.Run(t, new(CommonTestSuite))
 }

bcdaworker/queueing/manager/alr.go

@@ -166,10 +166,6 @@ func (q *masterQueue) startAlrJob(queJob *que.Job) error {
 		return err
 	}
 
-	if err := q.repository.IncrementCompletedJobCount(ctx, jobArgs.ID); err != nil {
-		q.alrLog.Warnf("Failed to increment completed count %s", err.Error())
-		return err
-	}
 	jobComplete, err := q.isJobComplete(ctx, jobArgs.ID)
 	if err != nil {
 		q.alrLog.Warnf("Failed to check job completion %s", err)
@@ -223,5 +219,6 @@ func (q *masterQueue) isJobComplete(ctx context.Context, jobID uint) (bool, erro
 			Println("Excess number of jobs completed.")
 		return true, nil
 	}
+
 	return false, nil
 }

bcdaworker/queueing/manager/que_test.go

@@ -213,8 +213,7 @@ func TestStartAlrJob(t *testing.T) {
 		Status:          models.JobStatusPending,
 		TransactionTime: time.Now(),
 		// JobCount is partitioned automatically, but it is done manually here
-		JobCount:          2,
-		CompletedJobCount: 0,
+		JobCount: 2,
 	}
 	id, err := r.CreateJob(ctx, job)
 	assert.NoError(t, err)

bcdaworker/repository/postgres/repository.go

@@ -85,7 +85,7 @@ func (r *Repository) GetCCLFBeneficiaryByID(ctx context.Context, id uint) (*mode
 
 func (r *Repository) GetJobByID(ctx context.Context, jobID uint) (*models.Job, error) {
 	sb := sqlFlavor.NewSelectBuilder()
-	sb.Select("id", "aco_id", "request_url", "status", "transaction_time", "job_count", "completed_job_count", "created_at", "updated_at")
+	sb.Select("id", "aco_id", "request_url", "status", "transaction_time", "job_count", "created_at", "updated_at")
 	sb.From("jobs").Where(sb.Equal("id", jobID))
 
 	query, args := sb.Build()
@@ -96,7 +96,7 @@ func (r *Repository) GetJobByID(ctx context.Context, jobID uint) (*models.Job, e
 	)
 
 	err := r.QueryRowContext(ctx, query, args...).Scan(&j.ID, &j.ACOID, &j.RequestURL, &j.Status, &transactionTime,
-		&j.JobCount, &j.CompletedJobCount, &createdAt, &updatedAt)
+		&j.JobCount, &createdAt, &updatedAt)
 	j.TransactionTime, j.CreatedAt, j.UpdatedAt = transactionTime.Time, createdAt.Time, updatedAt.Time
 
 	if err != nil {
@@ -121,27 +121,6 @@ func (r *Repository) UpdateJobStatusCheckStatus(ctx context.Context, jobID uint,
 		map[string]interface{}{"status": new})
 }
 
-func (r *Repository) IncrementCompletedJobCount(ctx context.Context, jobID uint) error {
-	ub := sqlFlavor.NewUpdateBuilder().Update("jobs")
-	ub.Set(ub.Incr("completed_job_count"), ub.Assign("updated_at", sqlbuilder.Raw("NOW()"))).
-		Where(ub.Equal("id", jobID))
-
-	query, args := ub.Build()
-	res, err := r.ExecContext(ctx, query, args...)
-	if err != nil {
-		return err
-	}
-	count, err := res.RowsAffected()
-	if err != nil {
-		return err
-	}
-	if count == 0 {
-		return fmt.Errorf("job %d not updated, no job found", jobID)
-	}
-
-	return nil
-}
-
 func (r *Repository) CreateJobKey(ctx context.Context, jobKey models.JobKey) error {
 	ib := sqlFlavor.NewInsertBuilder().InsertInto("job_keys")
 	ib.Cols("job_id", "que_job_id", "file_name", "resource_type").

bcdaworker/repository/postgres/repository_test.go

@@ -90,8 +90,8 @@ func (r *RepositoryTestSuite) TestJobsMethods() {
 	postgrestest.CreateACO(r.T(), r.db, aco)
 	defer postgrestest.DeleteACO(r.T(), r.db, aco.UUID)
 
-	failed := models.Job{ACOID: aco.UUID, Status: models.JobStatusFailed, CompletedJobCount: 1, TransactionTime: now}
-	completed := models.Job{ACOID: aco.UUID, Status: models.JobStatusCompleted, CompletedJobCount: 2, TransactionTime: now}
+	failed := models.Job{ACOID: aco.UUID, Status: models.JobStatusFailed, TransactionTime: now}
+	completed := models.Job{ACOID: aco.UUID, Status: models.JobStatusCompleted, TransactionTime: now}
 	postgrestest.CreateJobs(r.T(), r.db, &failed, &completed)
 
 	failed1, err := r.repository.GetJobByID(ctx, failed.ID)
@@ -117,12 +117,6 @@ func (r *RepositoryTestSuite) TestJobsMethods() {
 	failed.UpdatedAt = afterUpdate.UpdatedAt
 	assertJobsEqual(assert, failed, *afterUpdate)
 
-	assert.NoError(r.repository.IncrementCompletedJobCount(ctx, failed.ID))
-	afterUpdate, err = r.repository.GetJobByID(ctx, failed.ID)
-	assert.NoError(err)
-	assert.True(afterUpdate.UpdatedAt.After(failed.UpdatedAt))
-	assert.Equal(afterUpdate.CompletedJobCount, failed.CompletedJobCount+1)
-
 	// After all of these updates, the completed job should remain untouched
 	completed1, err := r.repository.GetJobByID(ctx, completed.ID)
 	assert.NoError(err)
@@ -141,10 +135,6 @@ func (r *RepositoryTestSuite) TestJobsMethods() {
 
 	err = r.repository.UpdateJobStatusCheckStatus(ctx, 0, models.JobStatusFailed, models.JobStatusArchived)
 	assert.EqualError(err, "job was not updated, no match found")
-
-	err = r.repository.IncrementCompletedJobCount(ctx, 0)
-	assert.EqualError(err, "job 0 not updated, no job found")
-
 }
 
 // TestJobKeysMethods validates the CRUD operations associated with the job_keys table

bcdaworker/repository/repository.go

@@ -31,8 +31,6 @@ type jobRepository interface {
 	// UpdateJobStatusCheckStatus updates the particular job indicated by the jobID
 	// iff the Job's status field matches current.
 	UpdateJobStatusCheckStatus(ctx context.Context, jobID uint, current, new models.JobStatus) error
-
-	IncrementCompletedJobCount(ctx context.Context, jobID uint) error
 }
 
 type jobKeyRepository interface {