Method Do

pkg/filemanager/manager/upload.go:501–548

(ctx context.Context)

}

func (m *UploadSentinelCheckTask) Do(ctx context.Context) (task.Status, error) {
	dep := dependency.FromContext(ctx)
	taskClient := dep.TaskClient()
	l := dep.Logger()
	fm := NewFileManager(dep, inventory.UserFromContext(ctx)).(*manager)

	// Check if sentinel is canceled due to callback complete
	t, err := taskClient.GetTaskByID(ctx, m.ID())
	if err != nil {
		return task.StatusError, fmt.Errorf("failed to get task by ID: %w", err)
	}

	if t.Status == task.StatusCompleted {
		l.Info("Upload sentinel check task [%d] is canceled due to callback complete.", m.ID())
		return task.StatusCompleted, nil
	}

	// unmarshal state
	state := &UploadSentinelCheckTaskState{}
	if err := json.Unmarshal([]byte(m.State()), state); err != nil {
		return task.StatusError, fmt.Errorf("failed to unmarshal state: %w", err)
	}

	l.Info("Upload sentinel check triggered, clean up stale place holder entity [%d].", state.Session.EntityID)
	entity, err := fm.fs.GetEntity(ctx, state.Session.EntityID)
	if err != nil {
		l.Debug("Failed to get entity [%d]: %s, skip sentinel check.", state.Session.EntityID, err)
		return task.StatusCompleted, nil
	}

	_, d, err := fm.getEntityPolicyDriver(ctx, entity, nil)
	if err != nil {
		l.Debug("Failed to get storage driver for entity [%d]: %s", state.Session.EntityID, err)
		return task.StatusError, err
	}

	_, err = d.Delete(ctx, entity.Source())
	if err != nil {
		l.Debug("Failed to delete entity source [%d]: %s", state.Session.EntityID, err)
		return task.StatusError, err
	}

	if err := d.CancelToken(ctx, state.Session); err != nil {
		l.Debug("Failed to cancel token [%d]: %s", state.Session.EntityID, err)
	}

	return task.StatusCompleted, nil
}