fix: use new job schema

ceramicnetwork · Oct 23, 2023 · 42f52ff · 42f52ff
1 parent 06b45ed
commit 42f52ff
Show file tree

Hide file tree

Showing 4 changed files with 31 additions and 29 deletions.
diff --git a/ci/cmd/deploy/main.go b/ci/cmd/deploy/main.go
@@ -61,6 +61,6 @@ func createJob(ctx context.Context) (string, error) {
 				return "", err
 			}
 		}
-		return newJob.Id, nil
+		return newJob.Job, nil
 	}
 }
diff --git a/common/aws/ddb/job.go b/common/aws/ddb/job.go
@@ -17,7 +17,7 @@ import (
 	"github.com/ceramicnetwork/go-cas/models"
 )
 
-const idTsIndex = "id-ts-index"
+const jobTsIndex = "job-ts-index"
 
 type JobDatabase struct {
 	ddbClient *dynamodb.Client
@@ -36,41 +36,42 @@ func NewJobDb(ctx context.Context, logger models.Logger, ddbClient *dynamodb.Cli
 
 func (jdb *JobDatabase) createJobTable(ctx context.Context) error {
 	createJobTableInput := dynamodb.CreateTableInput{
+		BillingMode: types.BillingModePayPerRequest,
 		AttributeDefinitions: []types.AttributeDefinition{
 			{
-				AttributeName: aws.String("stage"),
+				AttributeName: aws.String("id"),
 				AttributeType: "S",
 			},
 			{
-				AttributeName: aws.String("ts"),
-				AttributeType: "N",
+				AttributeName: aws.String("job"),
+				AttributeType: "S",
 			},
 			{
-				AttributeName: aws.String("id"),
+				AttributeName: aws.String("stage"),
 				AttributeType: "S",
 			},
-		},
-		KeySchema: []types.KeySchemaElement{
 			{
-				AttributeName: aws.String("stage"),
-				KeyType:       "HASH",
+				AttributeName: aws.String("type"),
+				AttributeType: "S",
 			},
 			{
 				AttributeName: aws.String("ts"),
-				KeyType:       "RANGE",
+				AttributeType: "N",
 			},
 		},
-		TableName: aws.String(jdb.jobTable),
-		ProvisionedThroughput: &types.ProvisionedThroughput{
-			ReadCapacityUnits:  aws.Int64(1),
-			WriteCapacityUnits: aws.Int64(1),
+		KeySchema: []types.KeySchemaElement{
+			{
+				AttributeName: aws.String("id"),
+				KeyType:       "HASH",
+			},
 		},
+		TableName: aws.String(jdb.jobTable),
 		GlobalSecondaryIndexes: []types.GlobalSecondaryIndex{
 			{
-				IndexName: aws.String(idTsIndex),
+				IndexName: aws.String(jobTsIndex),
 				KeySchema: []types.KeySchemaElement{
 					{
-						AttributeName: aws.String("id"),
+						AttributeName: aws.String("job"),
 						KeyType:       "HASH",
 					},
 					{
@@ -81,10 +82,6 @@ func (jdb *JobDatabase) createJobTable(ctx context.Context) error {
 				Projection: &types.Projection{
 					ProjectionType: types.ProjectionTypeAll,
 				},
-				ProvisionedThroughput: &types.ProvisionedThroughput{
-					ReadCapacityUnits:  aws.Int64(1),
-					WriteCapacityUnits: aws.Int64(1),
-				},
 			},
 		},
 	}
@@ -122,15 +119,15 @@ func (jdb *JobDatabase) CreateJob(ctx context.Context) (string, error) {
 		if err != nil {
 			return "", err
 		} else {
-			return newJob.Id, nil
+			return newJob.Job, nil
 		}
 	}
 }
 
 func (jdb *JobDatabase) QueryJob(ctx context.Context, id string) (*models.JobState, error) {
 	queryInput := dynamodb.QueryInput{
 		TableName:                 aws.String(jdb.jobTable),
-		IndexName:                 aws.String(idTsIndex),
+		IndexName:                 aws.String(jobTsIndex),
 		KeyConditionExpression:    aws.String("#id = :id"),
 		ExpressionAttributeNames:  map[string]string{"#id": "id"},
 		ExpressionAttributeValues: map[string]types.AttributeValue{":id": &types.AttributeValueMemberS{Value: id}},

diff --git a/models/job.go b/models/job.go
@@ -8,6 +8,7 @@ import (
 
 const DeployComponent = "casv5"
 const WorkerVersion = "5"
+const DefaultJobTtl = 2 * 7 * 24 * time.Hour
 
 type JobType string
 
@@ -42,19 +43,23 @@ const (
 )
 
 type JobState struct {
+	Job    string                 `dynamodbav:"job"`
 	Stage  JobStage               `dynamodbav:"stage"`
-	Ts     time.Time              `dynamodbav:"ts"`
-	Id     string                 `dynamodbav:"id"`
 	Type   JobType                `dynamodbav:"type"`
+	Ts     time.Time              `dynamodbav:"ts"`
 	Params map[string]interface{} `dynamodbav:"params"`
+	Id     string                 `dynamodbav:"id" json:"-"`
+	Ttl    time.Time              `dynamodbav:"ttl,unixtime" json:"-"`
 }
 
 func NewJob(jobType JobType, params map[string]interface{}) JobState {
 	return JobState{
+		Job:    uuid.New().String(),
 		Stage:  JobStage_Queued,
-		Ts:     time.Now(),
-		Id:     uuid.New().String(),
 		Type:   jobType,
+		Ts:     time.Now(),
 		Params: params,
+		Id:     uuid.New().String(),
+		Ttl:    time.Now().Add(DefaultJobTtl),
 	}
 }
diff --git a/services/test_helpers.go b/services/test_helpers.go
@@ -111,8 +111,8 @@ func (m *MockJobRepository) CreateJob(_ context.Context) (string, error) {
 		return "", fmt.Errorf("failed to create job")
 	}
 	newJob := models.NewJob(models.JobType_Anchor, nil)
-	m.jobStore[newJob.Id] = &newJob
-	return newJob.Id, nil
+	m.jobStore[newJob.Job] = &newJob
+	return newJob.Job, nil
 }
 
 func (m *MockJobRepository) QueryJob(_ context.Context, id string) (*models.JobState, error) {