cmsintflow

README

cmsintflow: QC/Conv Workflow State Machine

Overview

This package handles the QC (Quality Check) and Conversion workflow for PLATEAU CityGML data via FME Flow integration. The workflow is driven by CMS webhooks and Flow result callbacks.

Architecture

handler.go          req.go                 workflow.go              res.go
┌──────────┐    ┌───────────────┐    ┌──────────────────┐    ┌────────────────┐
│ Webhook  │───►│sendRequestTo  │───►│WorkflowMachine   │    │receiveResult   │
│          │    │Flow()         │    │ .Transition()    │    │FromFlow()      │
└──────────┘    │               │    │                  │    │                │
                │ Build SM      │    │ Pure logic:      │◄───│ Build SM       │
                │ Execute       │    │ - State          │    │ Execute        │
                │ Actions       │    │ - Config         │    │ Actions        │
                └───────────────┘    │ - Events→Actions │    └────────────────┘
                                     └──────────────────┘

The state machine (workflow.go) is pure — it contains no I/O. It takes the current item state and an event, and returns a list of actions. The callers (req.go, res.go) execute those actions (CMS updates, Flow API calls).

Item Fields

Field	CMS Key	Type	Values
QC Status	qc_status	tag (metadata)	nil, 未実行, 実行中, 成功, エラー
Conv Status	conv_status	tag (metadata)	nil, 未実行, 実行中, 成功, エラー
Skip QC/Conv	skip_qc_conv	tag (metadata)	See below

skip_qc_conv Values

Value	Meaning	skipQC	skipConv
nil (unset)	Run both QC and Conv	-	-
品質検査・変換を実行	Run both QC and Conv	false	false
品質検査のみ実行	Run QC only, skip Conv	false	true
変換のみ実行	Run Conv only, skip QC	true	false
品質検査のみをスキップ	Skip QC, run Conv	true	false
変換のみをスキップ	Skip Conv, run QC	false	true
品質検査・変換をスキップ	Skip both	true	true

State Machine

Types (workflow.go)

Phase:    idle | running | succeeded | failed
State:    (QC Phase, Conv Phase)
Config:   SkipQC, SkipConv, FeatureConv
Event:    WebhookReceived | QCCompleted(QCOK) | ConvCompleted | FlowFailed
Action:   Skip | SetStatus(qc, conv) | StartQC | StartConv | Fail(msg)

Transition Table

State (QC, Conv)	Event	Condition	Actions	New State
(idle, idle)	Webhook	!skipQC	SetStatus(qc=実行中, conv=未実行), StartQC	(running, idle)
(idle, idle)	Webhook	skipQC, !skipConv	SetStatus(conv=実行中), StartConv	(idle, running)
(any, any)	Webhook	skipQC && skipConv	Skip	—
(running, idle)	Webhook	skipQC, !skipConv	Skip (QC running guard)	—
(≠idle, any)	Webhook	!skipQC	Skip (idempotency)	—
(any, ≠idle)	Webhook	skipQC, !skipConv	Skip (idempotency)	—
(running, idle)	QCCompleted(ok)	!skipConv, featureConv	SetStatus(qc=成功), StartConv	(succeeded, running)
(running, idle)	QCCompleted(ok)	skipConv or !featureConv	SetStatus(qc=成功)	(succeeded, idle)
(running, idle)	QCCompleted(!ok)	—	SetStatus(qc=エラー)	(failed, idle)
(any, running)	ConvCompleted	—	SetStatus(conv=成功)	(any, succeeded)
(any, any)	FlowFailed	—	Fail(msg)	(running→failed, running→failed)

Normal Flow: QC + Conv

  (idle, idle)   ──Webhook──►  (running, idle)  ──QC OK──►  (succeeded, running)  ──Conv OK──►  (succeeded, succeeded)
                  StartQC                         StartConv                                       Done!

Conv Only Flow (skip_qc_conv = "変換のみ実行")

  (idle, idle)   ──Webhook──►  (idle, running)  ──Conv OK──►  (idle, succeeded)
                  StartConv                                     Done!

QC Failure

  (running, idle)  ──QC NG──►  (failed, idle)
                                Conv is NOT triggered.

Flow Failure (error during QC or Conv)

  (running, idle)   ──FlowFailed──►  (failed, idle)
  (any, running)    ──FlowFailed──►  (any, failed)

Key Design Decisions

1. Pure state machine: WorkflowMachine.Transition() has no side effects. All I/O (CMS updates, Flow API) is performed by the caller based on returned actions. This makes the state logic fully testable without mocks.
2. QC before Conv: When both are enabled, QC runs first. Conv is triggered by QCCompleted(ok=true) returning StartConv.
3. Idempotency: If a phase has already moved past idle, the webhook won't re-trigger it. This prevents re-entry from webhook cascades (CMS fires webhooks when status is updated).
4. QC running guard: When conv-only is requested but QC is still running, the webhook is skipped. Conv will be triggered by the QC result callback instead.
5. "実行" tag values: IsQCAndConvSkipped() (in cmsintegrationcommon) handles both "スキップ" and "実行" values in the skip_qc_conv tag.

Documentation

Index

• Variables
• func Handler(conf Config, g *echo.Group) error
• func ReqTypeForAction(a Action) cmsintegrationcommon.ReqType
• func WebhookHandler(conf Config) (cmswebhook.Handler, error)
• type Action
• type ActionKind
• type Config
• type Event
• type EventKind
• type Flow
  • func NewFlow(h *http.Client, baseURL, token string) Flow
• type FlowInternalResult
• type FlowRequest
  • func (r FlowRequest) MarshalJSON() ([]byte, error)
• type FlowRequestResult
• type FlowResult
  • func (r FlowResult) IDsMessage() string
  • func (r FlowResult) Internal() (res FlowInternalResult)
  • func (r FlowResult) IsFailed() bool
  • func (r FlowResult) IsSucceeded() bool
• type ID
  • func (i ID) Sign(secret string) string
• type PCMS
• type Phase
• type Services
  • func NewServices(c Config) (s *Services, _ error)
  • func (s *Services) ClearFlowRunID(ctx context.Context, itemID string) error
  • func (s *Services) DownloadAsset(ctx context.Context, assetID string) (io.ReadCloser, error)
  • func (s *Services) DownloadAssetAsBytes(ctx context.Context, assetID string) ([]byte, error)
  • func (s *Services) Fail(ctx context.Context, itemID string, ty cmsintegrationcommon.ReqType, ...) error
  • func (s *Services) GET(ctx context.Context, url string) (io.ReadCloser, error)
  • func (s *Services) GETAsBytes(ctx context.Context, url string) ([]byte, error)
  • func (s *Services) GetMainItemWithMetadata(ctx context.Context, i *cms.Item) (_ *cms.Item, err error)
  • func (s *Services) UpdateFeatureItemStatus(ctx context.Context, itemID string, ty cmsintegrationcommon.ReqType, ...) error
  • func (s *Services) UpdateFlowRunID(ctx context.Context, itemID, runID, triggerID string) error
  • func (s *Services) UpdateStatus(ctx context.Context, itemID string, ...) error
  • func (s *Services) UploadAsset(ctx context.Context, pid, url string) (_ string, err error)
• type WorkflowConfig
• type WorkflowMachine
  • func NewWorkflowMachine(item *cmsintegrationcommon.FeatureItem, featureQC, featureConv bool) *WorkflowMachine
  • func (m *WorkflowMachine) Transition(event Event) ([]Action, error)
• type WorkflowState

Variables

var ErrInvalidID = errors.New("invalid fme id")

Functions

func Handler

func Handler(conf Config, g *echo.Group) error

func ReqTypeForAction

func ReqTypeForAction(a Action) cmsintegrationcommon.ReqType

ReqTypeForAction returns the ReqType that should be used when executing an action. This bridges the state machine's action output to the existing Flow trigger API.

func WebhookHandler

func WebhookHandler(conf Config) (cmswebhook.Handler, error)

Types

type Action

type Action struct {
	Kind       ActionKind
	QCStatus   cmsintegrationcommon.ConvertionStatus // for ActionSetStatus
	ConvStatus cmsintegrationcommon.ConvertionStatus // for ActionSetStatus
	Message    string                                // for ActionFail
}

Action represents a side-effect to be executed by the caller. The caller interprets these to perform I/O (CMS updates, Flow API calls).

type ActionKind

type ActionKind int

ActionKind identifies a side-effect to be executed by the caller.

const (
	ActionSkip      ActionKind = iota // do nothing
	ActionSetStatus                   // update CMS item status fields
	ActionStartQC                     // trigger QC flow job
	ActionStartConv                   // trigger Conv flow job
	ActionFail                        // set error status + comment
)

type Config

type Config struct {
	Host             string
	CMSBaseURL       string
	CMSToken         string
	CMSSystemProject string
	CMSIntegration   string
	FlowBaseURL      string
	FlowToken        string
	Secret           string
}

type Event

type Event struct {
	Kind    EventKind
	QCOK    bool   // only for EventQCCompleted: true if QC passed without errors
	Message string // for EventFlowFailed: error message
}

Event represents something that happened in the system.

type EventKind

type EventKind int

EventKind identifies what happened.

const (
	EventWebhookReceived EventKind = iota // CMS webhook fired
	EventQCCompleted                      // QC job finished (check QCOK)
	EventConvCompleted                    // Conv job finished
	EventFlowFailed                       // Flow reported a failure
)

type Flow

type Flow interface {
	Request(context.Context, FlowRequest) (FlowRequestResult, error)
	Cancel(ctx context.Context, baseURL, runID, triggerID string) error
}

func NewFlow

func NewFlow(h *http.Client, baseURL, token string) Flow

type FlowInternalResult

type FlowInternalResult struct {
	Conv     map[string][]string
	Dic      string
	QCResult string
	QCOK     bool
}

type FlowRequest

type FlowRequest struct {
	TriggerID       string
	BaseURL         string // Optional: overrides client's base URL if set
	NotificationURL string
	AuthToken       string
	CityGMLURL      string
	ConvSettings    *cmsintegrationcommon.ConvSettings
	DryRun          bool
}

func (FlowRequest) MarshalJSON

func (r FlowRequest) MarshalJSON() ([]byte, error)

type FlowRequestResult

type FlowRequestResult map[string]any

type FlowResult

type FlowResult struct {
	ID           string   `json:"-"`
	RunID        string   `json:"runId"`
	TriggerID    string   `json:"triggerId"`
	DeploymentID string   `json:"deploymentId"`
	Status       string   `json:"status"`
	Logs         []string `json:"logs"`
	Outputs      []string `json:"outputs"`
}

func (FlowResult) IDsMessage

func (r FlowResult) IDsMessage() string

func (FlowResult) Internal

func (r FlowResult) Internal() (res FlowInternalResult)

Internal parses Flow outputs and extracts conversion results, dictionary, and QC results

func (FlowResult) IsFailed

func (r FlowResult) IsFailed() bool

func (FlowResult) IsSucceeded

func (r FlowResult) IsSucceeded() bool

type ID

type ID struct {
	ItemID      string
	ProjectID   string
	FeatureType string
	Type        cmsintegrationcommon.ReqType
}

func (ID) Sign

func (i ID) Sign(secret string) string

type PCMS

type PCMS interface {
	plateaucms.FeatureTypeStore
	plateaucms.MetadataStore
	plateaucms.SpecStore
}

type Phase

type Phase string

Phase represents the lifecycle phase of a single operation (QC or Conv).

const (
	PhaseIdle      Phase = "idle"      // not started (nil, 未実行)
	PhaseRunning   Phase = "running"   // currently executing (実行中)
	PhaseSucceeded Phase = "succeeded" // completed successfully (成功)
	PhaseFailed    Phase = "failed"    // completed with error (エラー)
)

type Services

type Services struct {
	CMS        cms.Interface
	HTTP       *http.Client
	TaskRunner gcptaskrunner.TaskRunner
	PCMS       PCMS
	Flow       Flow
}

func NewServices

func NewServices(c Config) (s *Services, _ error)

func (*Services) ClearFlowRunID

func (s *Services) ClearFlowRunID(ctx context.Context, itemID string) error

func (*Services) DownloadAsset

func (s *Services) DownloadAsset(ctx context.Context, assetID string) (io.ReadCloser, error)

func (*Services) DownloadAssetAsBytes

func (s *Services) DownloadAssetAsBytes(ctx context.Context, assetID string) ([]byte, error)

func (*Services) Fail

func (s *Services) Fail(ctx context.Context, itemID string, ty cmsintegrationcommon.ReqType, message string, args ...any) error

func (*Services) GET

func (s *Services) GET(ctx context.Context, url string) (io.ReadCloser, error)

func (*Services) GETAsBytes

func (s *Services) GETAsBytes(ctx context.Context, url string) ([]byte, error)

func (*Services) GetMainItemWithMetadata

func (s *Services) GetMainItemWithMetadata(ctx context.Context, i *cms.Item) (_ *cms.Item, err error)

func (*Services) UpdateFeatureItemStatus

func (s *Services) UpdateFeatureItemStatus(ctx context.Context, itemID string, ty cmsintegrationcommon.ReqType, status cmsintegrationcommon.ConvertionStatus) error

func (*Services) UpdateFlowRunID

func (s *Services) UpdateFlowRunID(ctx context.Context, itemID, runID, triggerID string) error

func (*Services) UpdateStatus

func (s *Services) UpdateStatus(ctx context.Context, itemID string, qcStatus, convStatus cmsintegrationcommon.ConvertionStatus) error

UpdateStatus updates the QC/Conv status fields directly. This is used by the workflow state machine's action execution.

func (*Services) UploadAsset

func (s *Services) UploadAsset(ctx context.Context, pid, url string) (_ string, err error)

type WorkflowConfig

type WorkflowConfig struct {
	SkipQC      bool // QC should be skipped (from IsQCAndConvSkipped or !featureType.QC)
	SkipConv    bool // Conv should be skipped (from IsQCAndConvSkipped or !featureType.Conv)
	FeatureConv bool // feature type supports Conv
}

WorkflowConfig captures static configuration that affects transitions.

type WorkflowMachine

type WorkflowMachine struct {
	State  WorkflowState
	Config WorkflowConfig
}

WorkflowMachine is a pure state machine for the QC/Conv workflow. It contains no I/O — the caller executes the returned actions.

func NewWorkflowMachine

func NewWorkflowMachine(item *cmsintegrationcommon.FeatureItem, featureQC, featureConv bool) *WorkflowMachine

NewWorkflowMachine builds the state machine from the current CMS item state.

func (*WorkflowMachine) Transition

func (m *WorkflowMachine) Transition(event Event) ([]Action, error)

Transition computes the next state and actions to execute.

type WorkflowState

type WorkflowState struct {
	QC   Phase
	Conv Phase
}

WorkflowState is the compound state of (qc_status, conv_status).