feat: add resource reconcile

server/api/errors.go

@@ -20,8 +20,9 @@ var (
 	ErrResourceNotFound = newErrorMessage(http.StatusNotFound, 40402, "resource not found")
 	DocumentNotFound    = newErrorMessage(http.StatusNotFound, 40403, "data not found in database")
 
-	ErrInternal = newErrorMessage(http.StatusInternalServerError, 50001, "internal error")
-	ErrUnknown  = newErrorMessage(http.StatusInternalServerError, 50002, "unknown error")
+	ErrInternal      = newErrorMessage(http.StatusInternalServerError, 50001, "internal error")
+	ErrUnknown       = newErrorMessage(http.StatusInternalServerError, 50002, "unknown error")
+	ErrResourceRetry = newErrorMessage(http.StatusInternalServerError, 50003, "resource retry")
 )
 
 func IsErrNotFound(err error) bool {

server/db/mongo.go

@@ -16,6 +16,7 @@ import (
 
 // below are general available
 const (
+	CollTools           = "tools"
 	CollToolSpec        = "tool_specs"
 	CollToolInstance    = "tool_instances"
 	CoolDistributedLock = "distributed_lock"

server/model/base.go

@@ -0,0 +1,51 @@
+package model
+
+import (
+	"context"
+	"time"
+
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
+
+func NewBase(ctx context.Context) Base {
+	return Base{
+		ID:        primitive.NewObjectID(),
+		CreatedAt: time.Now(),
+		UpdatedAt: time.Now(),
+	}
+}
+
+type Base struct {
+	ID        primitive.ObjectID `json:"id" bson:"_id"`
+	CreatedAt time.Time          `json:"created_at" bson:"created_at"`
+	UpdatedAt time.Time          `json:"updated_at" bson:"updated_at"`
+	CreatedBy string             `json:"created_by" bson:"created_by"`
+	UpdatedBy string             `json:"updated_by" bson:"updated_by"`
+}
+
+func (b Base) ObjectID() primitive.ObjectID {
+	return b.ID
+}
+
+func NewBaseResource(ctx context.Context) BaseResource {
+	b := Base{
+		ID:        primitive.NewObjectID(),
+		CreatedAt: time.Now(),
+		UpdatedAt: time.Now(),
+	}
+	return BaseResource{
+		Base:    b,
+		RetryAt: b.CreatedAt,
+	}
+}
+
+type BaseResource struct {
+	Base    `json:",inline" bson:",inline"`
+	Times   int       `json:"-" bson:"retry_times"`
+	RetryAt time.Time `json:"-" bson:"retry_at"`
+	Err     string    `json:"-" bson:"error,omitempty"`
+}
+
+func (b BaseResource) RetryTimes() int {
+	return b.Times
+}

server/model/resource.go

@@ -0,0 +1,18 @@
+package model
+
+type ResourceStatus string
+
+const (
+	ResourceStatusDraft      = ResourceStatus("draft")
+	ResourceStatusQueued     = ResourceStatus("queued")
+	ResourceStatusUpdated    = ResourceStatus("updated")
+	ResourceStatusProcessing = ResourceStatus("processing")
+	ResourceStatusRetry      = ResourceStatus("retry")
+	ResourceStatusSuccess    = ResourceStatus("success")
+	ResourceStatusToBeDelete = ResourceStatus("to_be_deleted")
+	ResourceStatusDeleting   = ResourceStatus("deleting")
+	ResourceStatusDeleted    = ResourceStatus("deleted")
+	ResourceStatusError      = ResourceStatus("error")
+	ResourceStatusBuilt      = ResourceStatus("built")
+	ResourcePublishing       = ResourceStatus("publishing")
+)

server/model/resource/tool.go

@@ -0,0 +1,7 @@
+package resource
+
+import "github.com/npi-ai/npi/server/model"
+
+type ToolResource struct {
+	model.BaseResource `json:",inline" bson:",inline"`
+}

server/reconcile/controller.go

@@ -0,0 +1,184 @@
+package reconcile
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/npi-ai/npi/server/api"
+	"github.com/npi-ai/npi/server/log"
+	"github.com/npi-ai/npi/server/model"
+	"github.com/npi-ai/npi/server/model/resource"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+type Resource interface {
+	resource.ToolResource
+	ObjectID() primitive.ObjectID
+	RetryTimes() int
+}
+
+type Controller[R Resource] interface {
+	Start(ctx context.Context) error
+}
+
+type BaseController[R Resource] struct {
+	name          string
+	maxGoroutines int
+}
+
+func (c *BaseController[Resource]) SetName(name string) {
+	c.name = name
+}
+
+func (c *BaseController[Resource]) SetMaximumParallelism(max int) {
+	c.maxGoroutines = max
+}
+
+func (c *BaseController[Resource]) GetMaximumParallelism() int {
+	if c.maxGoroutines == 0 {
+		return 64
+	}
+	return c.maxGoroutines
+}
+
+func (c *BaseController[Resource]) Reconcile(
+	ctx context.Context,
+	coll *mongo.Collection,
+	currentStatus, midStatus, expectedStatus model.ResourceStatus,
+	g func() Resource,
+	f func(ctx context.Context,
+		resource Resource) error) {
+	barrier := make(chan struct{}, c.GetMaximumParallelism())
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			opts := options.FindOneAndUpdateOptions{
+				Sort: bson.M{"retry_at": 1},
+			}
+			result := coll.FindOneAndUpdate(ctx,
+				bson.M{"status": currentStatus, "retry_at": bson.M{"$lte": time.Now()}},
+				bson.M{"$set": bson.M{
+					"status": midStatus,
+				}},
+				&opts,
+			)
+			if result.Err() != nil {
+				if !errors.Is(result.Err(), mongo.ErrNoDocuments) {
+					log.Error(ctx).Err(result.Err()).
+						Str("status", string(currentStatus)).
+						Str("resource", c.name).
+						Msg("failed to get object from database")
+					time.Sleep(10 * time.Second)
+				}
+				time.Sleep(time.Second)
+				continue
+			}
+			R := g()
+			err := result.Decode(&R)
+			if err != nil {
+				row, _err := result.Raw()
+				// TODO update error
+				log.Error(ctx).
+					Err(err).
+					Str("status", string(currentStatus)).
+					Str("resource", c.name).
+					Interface("row_err", _err).
+					Stringer("resource", row).Msg("failed to decode resource")
+				continue
+			}
+
+			barrier <- struct{}{}
+			go func(r Resource) {
+				defer func() {
+					<-barrier
+				}()
+				id := r.ObjectID()
+				start := time.Now()
+				log.Info(ctx).
+					Str("resource_id", id.Hex()).
+					Str("status", string(currentStatus)).
+					Str("resource", c.name).
+					Msg("start to process the resource")
+				if err = f(ctx, r); err != nil {
+					if api.IsError(err, api.ErrResourceRetry) {
+						log.Info(ctx).
+							Str("resource_id", r.ObjectID().Hex()).
+							Str("status", string(currentStatus)).
+							Str("resource", c.name).
+							Msg("the resource is not ready")
+						_, err = coll.UpdateOne(ctx,
+							bson.M{"_id": r.ObjectID()},
+							bson.M{"$set": bson.M{
+								"updated_at":  time.Now(),
+								"retry_at":    time.Now().Add(10 * time.Second),
+								"retry_times": r.RetryTimes() + 1,
+								"status":      currentStatus,
+							}},
+						)
+						if err != nil {
+							log.Error(ctx).Err(err).
+								Str("resource_id", r.ObjectID().Hex()).
+								Str("status", string(currentStatus)).
+								Str("resource", c.name).
+								Interface("resource", r).
+								Int("retry_times", r.RetryTimes()).
+								Msg("failed to update status")
+						}
+						return
+					}
+					log.Warn(ctx).Err(err).
+						Str("resource_id", id.Hex()).
+						Str("status", string(currentStatus)).
+						Str("resource", c.name).
+						Msg("failed to handle status")
+					_, err = coll.UpdateOne(ctx,
+						bson.M{"_id": id},
+						bson.M{"$set": bson.M{
+							"updated_at": time.Now(),
+							"status":     model.ResourceStatusError,
+							"error":      err.Error(),
+						}},
+					)
+					if err != nil {
+						log.Error(ctx).Err(err).
+							Str("resource_id", id.Hex()).
+							Str("status", string(currentStatus)).
+							Str("resource", c.name).
+							Interface("resource", r).
+							Msg("failed to update status")
+					}
+				} else {
+					log.Info(ctx).
+						Str("resource_id", id.Hex()).
+						Str("before", string(currentStatus)).
+						Str("after", string(expectedStatus)).
+						Str("resource", c.name).
+						Dur("duration", time.Since(start)).
+						Msg("the resource is handled successfully")
+					_, err = coll.UpdateOne(ctx,
+						bson.M{"_id": id},
+						bson.M{"$set": bson.M{
+							"updated_at": time.Now(),
+							"status":     expectedStatus,
+						}},
+					)
+					if err != nil {
+						log.Warn(ctx).
+							Str("resource_id", id.Hex()).
+							Str("before", string(currentStatus)).
+							Str("after", string(expectedStatus)).
+							Str("resource", c.name).
+							Dur("duration", time.Since(start)).
+							Msg("failed to update database")
+					}
+				}
+			}(R)
+		}
+	}
+}

server/reconcile/resource/workspace.go

@@ -0,0 +1,54 @@
+package resource
+
+import (
+	"context"
+	"github.com/npi-ai/npi/server/db"
+	"github.com/npi-ai/npi/server/log"
+	"github.com/npi-ai/npi/server/model"
+	"github.com/npi-ai/npi/server/model/resource"
+	"github.com/npi-ai/npi/server/reconcile"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+type toolController struct {
+	reconcile.BaseController[resource.ToolResource]
+	coll *mongo.Collection
+}
+
+func NewToolController() reconcile.Controller[resource.ToolResource] {
+	return &toolController{
+		coll: db.GetCollection(db.CollTools),
+	}
+}
+
+func (pr *toolController) Start(ctx context.Context) error {
+	pr.BaseController.SetName("Project")
+	g := func() resource.ToolResource {
+		return resource.ToolResource{}
+	}
+	result, err := pr.coll.UpdateMany(
+		ctx,
+		bson.M{"status": model.ResourceStatusProcessing},
+		bson.M{"$set": bson.M{"status": model.ResourceStatusUpdated}},
+	)
+	if err != nil {
+		return err
+	}
+
+	log.Info().Int("recovered", int(result.ModifiedCount)).Msg("tool recovered")
+	go pr.Reconcile(ctx, pr.coll,
+		model.ResourceStatusUpdated,
+		model.ResourceStatusProcessing,
+		model.ResourceStatusSuccess,
+		g, pr.UpdatedHandler)
+	return nil
+}
+
+func (pr *toolController) UpdatedHandler(ctx context.Context, ws resource.ToolResource) error {
+	return nil
+}
+
+func (pr *toolController) DeletingHandler(ctx context.Context, ws resource.ToolResource) error {
+	return nil
+}