Plugin 系统

// internal/plugin/callbuilder/types/types.go
type CallBuilder interface {
	Name() CallBuilderName
	Build(ctx context.Context, taskCtx *commontypes.TaskCtx, io TaskIOReader) (CallList, error)
}

type StreamingCallBuilder interface {
	CallBuilder
	PrepareStream(ctx context.Context, taskCtx *commontypes.TaskCtx, io TaskIOReader) (CallStreamProducer, error)
}

type CallStreamProducer interface {
	Run(ctx context.Context, emit BatchEmit) error
}
type BatchEmit func(ctx context.Context, batch CallList) error

// internal/plugin/event/types/types.go
type WriterPlugin interface {
	Name() PluginName
	Execute(ctx context.Context, taskCtx *commontypes.TaskCtx, outputs []any, io TaskIO) (any, error)
}

package mybuilder

import (
	"context"
	"fmt"

	"github.com/bytedance/sonic"

	commontypes "github.com/Chaintable/blockx/internal/common/types"
	"github.com/Chaintable/blockx/internal/plugin/callbuilder/types"
)

var jsonAPI = sonic.Config{UseNumber: true}.Froze()

type decl struct {
	Function string `json:"function"`
	Rows     []any  `json:"rows"`
}

// BuilderMyScan 需先加到 internal/plugin/callbuilder/types/types.go 的 CallBuilderName 常量块。
type Builder struct{}

func (b *Builder) Name() types.CallBuilderName { return types.BuilderMyScan }

func (b *Builder) Build(ctx context.Context, taskCtx *commontypes.TaskCtx, io types.TaskIOReader) (types.CallList, error) {
	var d decl
	if err := jsonAPI.Unmarshal(taskCtx.FunctionCallConfig.Config, &d); err != nil {
		return nil, fmt.Errorf("myScan: unmarshal config: %w", err)
	}
	calls := make(types.CallList, 0, len(d.Rows))
	for i, row := range d.Rows {
		c, err := types.CompactCall(d.Function, nil, []any{row}, fmt.Sprintf("%s-%d", taskCtx.ID, i))
		if err != nil {
			return nil, err
		}
		calls = append(calls, c)
	}
	return calls, nil
}

package myplugin

import (
	"context"

	commontypes "github.com/Chaintable/blockx/internal/common/types"
	"github.com/Chaintable/blockx/internal/plugin/event/types"
)

// PluginMyWrite 需先加到 internal/plugin/event/types/types.go 的 PluginName 常量块。
type Plugin struct{}

func (p *Plugin) Name() types.PluginName { return types.PluginMyWrite }

func (p *Plugin) Execute(ctx context.Context, taskCtx *commontypes.TaskCtx, outputs []any, io types.TaskIO) (any, error) {
	if taskCtx.ResultHandler == nil {
		return nil, &types.PluginError{Kind: types.PluginErrExecution, PluginName: p.Name(), Message: "result handler config is nil"}
	}
	rowCount, err := types.VisitOutputRows(outputs, func(row map[string]any) error {
		// 把 row 投影成写请求；不要保留 row map 本身
		return nil
	})
	if err != nil {
		return nil, &types.PluginError{Kind: types.PluginErrExecution, PluginName: p.Name(), Message: err.Error()}
	}
	// 写 IO 走传入的 io.Write；下游错误保留可重试分类：
	// if _, err := io.Write(ctx, req); err != nil {
	//     return nil, &types.PluginError{Kind: types.PluginErrExecution, PluginName: p.Name(),
	//         Message: "write failed", Cause: err, Retryable: types.IsRetryableError(err)}
	// }
	return map[string]any{"written_rows": rowCount}, nil
}

cd ~/Work/blockx
go test ./internal/plugin/... -short              # builder / plugin / operator 单测与包内集成
go test ./internal/worker/app/ -run 'Registry|Profile'   # Profile 装配
go test ./internal/worker/adapters/ -run 'Orchestrator_EventPhase|OrchestratorStream|BuilderFails'
go test ./internal/worker/devstub/