重构: 各类型 Controller 接口精简,配合 framework-v2 internal/handler 收拢公共实现

app.go/condflow.go/cron.go/event_stream.go/executor.go/gobridge.go/grpc.go/job.go/mcp.go/mqtt.go/res_virtual_table.go/ws.go/controller.go/support/service.go 精简掉搬到 framework-v2 internal/handler 里的公共默认实现,只保留接口定义。

app.go

@@ -3,7 +3,7 @@ package contracts
 type App interface {
 	LoadGoPlugin(file string) error
 	RegControllers(items map[string]Controller) error
-	RegListens(items map[string]ResListener) error
+	RegResWatchers(items map[string]ResWatcher) error
 	RegGRpcs(items map[string]GRpc) error
 	RegJobs(items map[string]Job) error
 	RegCrons(items map[string]Cron) error
@@ -12,7 +12,7 @@ type App interface {
 	RegGoBridges(items map[string]GoBridge) error
 	RegResVirtualTables(items map[string]ResVirtualTable) error
 
-	GetResListener(code string) (ResListener, bool)
+	GetResWatcher(code string) (ResWatcher, bool)
 	GetAppController(code string) (Controller, bool)
 	GetAppGRpc(code string) (GRpc, bool)
 	GetAppJob(code string) (Job, bool)

condflow.go

@@ -3,7 +3,6 @@ package contracts
 import (
 	"git.fsdpf.net/go/reflux"
 	"git.fsdpf.net/go/req"
-	"github.com/samber/do/v2"
 )
 
 type CondflowService interface {
@@ -14,19 +13,3 @@ type FlowDecision interface {
 	Controller
 	Res() req.Resource
 }
-
-type BaseFlowDecision struct {
-	Controller
-	res string
-}
-
-func (be BaseFlowDecision) Res() req.Resource {
-	return do.MustInvoke[MustResource](be.Container())(be.res)
-}
-
-func NewBaseFlowDecision(container do.Injector, res string) *BaseFlowDecision {
-	return &BaseFlowDecision{
-		Controller: &BaseController{container},
-		res:        res,
-	}
-}

controller.go

@@ -14,22 +14,3 @@ type Controller interface {
 	Dispatch(job string, payload any, u req.User) error
 }
 
-type BaseController struct {
-	container do.Injector
-}
-
-func (this BaseController) Container() do.Injector {
-	return this.container
-}
-
-func (BaseController) Init() error {
-	return nil
-}
-
-func (this BaseController) Dispatch(job string, payload any, u req.User) error {
-	return do.MustInvoke[JobService](this.Container()).Dispatch(job, payload, u)
-}
-
-func NewBaseController(container do.Injector) Controller {
-	return &BaseController{container}
-}

cron.go

@@ -14,10 +14,3 @@ type Cron interface {
 	Run(req.User) error
 }
 
-type CronBase struct {
-	Controller
-}
-
-func (CronBase) Run(req.User) error {
-	return nil
-}

event_stream.go

@@ -1,11 +1,9 @@
 package contracts
 
 import (
-	"io"
 	"net/http"
 
 	"git.fsdpf.net/go/req"
-	"github.com/samber/do/v2"
 )
 
 type EventStream interface {
@@ -21,47 +19,3 @@ type EventStream interface {
 	// 发送数据
 	Sent(data, event string) error
 }
-
-type EsHandleController struct {
-	Controller
-	w       http.ResponseWriter
-	flusher http.Flusher
-	request *http.Request
-}
-
-func (this EsHandleController) Request() *http.Request {
-	return this.request
-}
-
-func (this EsHandleController) Route() req.Route {
-	return this.Request().Context().Value(req.RouteCtx{Name: "Route"}).(req.Route)
-}
-
-func (this EsHandleController) User() req.User {
-	return this.Request().Context().Value(req.RouteCtx{Name: "User"}).(req.User)
-}
-
-func (this EsHandleController) Execute(req.GlobalParams) error {
-	return ErrFuncNotImplemented
-}
-
-func (this EsHandleController) Sent(data, event string) error {
-	if _, err := io.WriteString(this.w, "event: "+event+"\ndata: "+data+"\n\n"); err != nil {
-		return err
-	}
-	this.flusher.Flush()
-	return nil
-}
-
-func NewEventStreamController(container do.Injector, request *http.Request, w http.ResponseWriter) EventStream {
-	w.Header().Set("Content-Type", "text/event-stream")
-	w.Header().Set("Cache-Control", "no-cache")
-	w.Header().Set("Connection", "keep-alive")
-
-	return &EsHandleController{
-		Controller: &BaseController{container},
-		request:    request,
-		w:          w,
-		flusher:    w.(http.Flusher),
-	}
-}

executor.go

@@ -2,26 +2,9 @@ package contracts
 
 import (
 	"git.fsdpf.net/go/req"
-	"github.com/samber/do/v2"
 )
 
 type Executor interface {
 	Controller
 	Res() req.Resource
 }
-
-type BaseExecutor struct {
-	Controller
-	res string
-}
-
-func (be BaseExecutor) Res() req.Resource {
-	return do.MustInvoke[MustResource](be.Container())(be.res)
-}
-
-func NewBaseExecutor(container do.Injector, res string) *BaseExecutor {
-	return &BaseExecutor{
-		Controller: &BaseController{container},
-		res:        res,
-	}
-}

gobridge.go

@@ -9,14 +9,5 @@ type GoBridgeService interface {
 // GoBridge 继承 Controller，可通过 gobridge.WithHandlers 绑定到 Pool，
 // 所有公开方法自动暴露给 Python 通过 call_go() 调用。
 type GoBridge interface {
-	_gobridge_handler()
 	Controller
 }
-
-// GoBridgeBase 嵌入到用户 struct 以实现 GoBridge 接口
-type GoBridgeBase struct {
-	Controller
-}
-
-// _gobridge_handler 签名函数
-func (bb *GoBridgeBase) _gobridge_handler() {}

grpc.go

@@ -16,10 +16,3 @@ type GRpc interface {
 	GetGRpcServiceDesc() *grpc.ServiceDesc
 }
 
-type GRpcBase struct {
-	Controller
-}
-
-func (GRpcBase) GetGRpcServiceDesc() *grpc.ServiceDesc {
-	return nil
-}

job.go

@@ -13,12 +13,18 @@ type Job interface {
 	Controller
 	// 任务处理
 	Handle(any, req.User) error
+	// OnExceeded 重试次数耗尽时由框架回调，用于清理状态。默认实现为空操作。
+	OnExceeded(any, req.User) error
 }
 
-type JobBase struct {
+// NoRetryError 表示不可重试的错误，job 返回此类型后框架立即标记为 exceeded，不再调度重试。
-	Controller
+type NoRetryError struct {
+	Err error
 }
 
-func (JobBase) Handle(any, req.User) error {
+func (e *NoRetryError) Error() string { return e.Err.Error() }
-	return nil
+func (e *NoRetryError) Unwrap() error { return e.Err }
-}
+
+// ErrNoRetry 将 err 包装为不可重试错误，job handler 使用此函数终止重试。
+func ErrNoRetry(err error) error { return &NoRetryError{Err: err} }
+

mcp.go

@@ -1,7 +1,76 @@
 package contracts
 
-import "net/http"
+import (
+	"context"
+
+	"git.fsdpf.net/go/req"
+	"github.com/mark3labs/mcp-go/mcp"
+	"github.com/mark3labs/mcp-go/server"
+	openai "github.com/sashabaranov/go-openai"
+)
+
+// InjectOption 是 InjectTools 的可选配置项。
+type InjectOption func(*InjectOptions)
+
+type InjectOptions struct {
+	Codes    []string
+	OnCall   func(openai.ToolCall)         // 工具执行前回调
+	OnResult func(openai.ToolCall, string) // 工具执行后回调
+}
+
+// WithFilter 只注入指定 apiCode 的 MCP server 工具；不传则注入全部已注册的。
+func WithFilter(codes ...string) InjectOption {
+	return func(o *InjectOptions) { o.Codes = codes }
+}
+
+// WithOnCall 设置工具执行前回调，tc 为完整的 ToolCall 对象。
+func WithOnCall(fn func(tc openai.ToolCall)) InjectOption {
+	return func(o *InjectOptions) { o.OnCall = fn }
+}
+
+// WithOnResult 设置工具执行后回调，tc 为 ToolCall 对象，result 为返回文本。
+func WithOnResult(fn func(tc openai.ToolCall, result string)) InjectOption {
+	return func(o *InjectOptions) { o.OnResult = fn }
+}
 
 type MCPService interface {
-	ServeHTTP(w http.ResponseWriter, r *http.Request)
+	Serve(req.Route) MCPServer
+	Shutdown() error
+	// InjectTools 将工具列表注入 ChatCompletionRequest.Tools，同时返回 Instructions 和工具调用执行器。
+	// ctx 须携带 req.User（WithToolFilter 据此过滤工具）。
+	// 通过 WithOnCall / WithOnResult 设置实时进度回调，WithFilter 限定注入范围。
+	InjectTools(ctx context.Context, impl mcp.Implementation, req *openai.ChatCompletionRequest, opts ...InjectOption) (
+		instructions string,
+		handleToolCalls func(toolCalls []openai.ToolCall) []openai.ChatCompletionMessage,
+		err error,
+	)
+}
+
+type MCPServer interface {
+	Server
+	Shutdown() error
+}
+
+type MCPTool interface {
+	Controller
+	// 获取 MCPServer
+	MCP() *server.MCPServer
+	// Sent 在工具执行期间通过 notifications/progress 向客户端推送进度消息。
+	// 客户端请求须携带 _meta.progressToken，否则本方法为空操作。
+	Sent(ctx context.Context, data string) error
+	// Execute 处理请求并返回响应数据，返回值支持以下类型：
+	//   - mcp.Content / []mcp.Content    直接作为工具内容返回
+	//   - *mcp.CallToolResult            完整结果，直接透传
+	//   - contracts.Errno / *contracts.Errno  业务错误码，转为工具错误
+	//   - contracts.Err / *contracts.Err      同上
+	//   - string                         文本结果
+	//   - []byte                         文本结果（原始字节）
+	//   - error                          工具错误
+	//   - nil                            空结果
+	//   - 其他任意类型                    JSON 序列化后返回，同时填充 StructuredContent
+	Execute(context.Context, req.GlobalParams) any
+}
+
+type MCPResource interface {
+	Controller
 }

mqtt.go

@@ -2,7 +2,6 @@ package contracts
 
 import (
 	"git.fsdpf.net/go/req"
-	"github.com/samber/do/v2"
 )
 
 type MqttService interface {
@@ -56,40 +55,3 @@ type Mqtt interface {
 	// OnRetainedExpired(g GlobalParams, topic string) error
 }
 
-type MqttHandle struct {
-	Controller
-}
-
-func (this *MqttHandle) OnConnect(g req.GlobalParams) error {
-	return ErrFuncNotImplemented
-}
-
-func (this *MqttHandle) OnDisconnect(g req.GlobalParams) error {
-	return ErrFuncNotImplemented
-}
-
-func (this *MqttHandle) OnSubscribed(g req.GlobalParams, topic string) error {
-	return ErrFuncNotImplemented
-}
-
-func (this MqttHandle) OnUnsubscribed(g req.GlobalParams, topic string) error {
-	return ErrFuncNotImplemented
-}
-
-func (this MqttHandle) OnMessage(g req.GlobalParams, topic string, retain bool, qos byte) error {
-	return ErrFuncNotImplemented
-}
-
-func (this MqttHandle) OnMessageDropped(g req.GlobalParams, topic string, retain bool, qos byte) error {
-	return ErrFuncNotImplemented
-}
-
-func (this MqttHandle) OnPublish(g req.GlobalParams, topic string, retain bool, qos byte, raw []byte) ([]byte, error) {
-	return nil, ErrFuncNotImplemented
-}
-
-func NewMqttController(container do.Injector) Mqtt {
-	return &MqttHandle{
-		Controller: &BaseController{container},
-	}
-}

res_virtual_table.go

@@ -1,99 +1,16 @@
 package contracts
 
 import (
-	"git.fsdpf.net/go/db"
-	"git.fsdpf.net/go/reflux"
 	"git.fsdpf.net/go/req"
-	"github.com/samber/do/v2"
 )
 
-// ResVirtualTable 虚拟表接口，由业务层实现，框架通过 SQLite vtab 机制调用。
+// ResVirtualTableSetup 由 framework-v2 在 sqlite_vtable build tag 下注册到容器，负责向 SQLite
-// Select 提供列表查询，Fetch 提供单条完整查询（含 Detail 补充字段），
+// 注册虚拟表模块，具体说明见 req.ResVirtualTableSetup。
-// Insert/Update/Delete 支持写操作并自动同步缓存。
+type ResVirtualTableSetup = req.ResVirtualTableSetup
+
+// ResVirtualTable 在 req.ResVirtualTable 基础上合并 Controller，使实现方可以访问 DI 容器等框架能力；
+// 具体方法说明见 req.ResVirtualTable。
 type ResVirtualTable interface {
 	Controller
-	GetResource() req.Resource
+	req.ResVirtualTable
-
-	// Fetch 按主键获取单条完整数据，内部负责调用远端接口并补充 Detail 专属字段。
-	// 框架在 item cache miss 时调用，结果写入 item cache 供后续查询复用。
-	Fetch(pk any) (map[string]any, error)
-
-	// Detail 对已有的 item 数据进行补充，填充列表接口不返回的详情字段。
-	// changed=true 时框架会将修改后的数据回写 item cache。
-	Detail(item reflux.R) (changed bool, err error)
-
-	// Select 查询列表数据，filter 为过滤条件，pagesize/page 控制分页。
-	// 返回的 items 每项为 map[string]any，total 为总记录数。
-	Select(filter reflux.R, pagesize, page int) (items []any, total int64, err error)
-
-	// Insert 新增一条记录，返回新记录的 rowid。
-	Insert(item reflux.R) (rowid int64, err error)
-
-	// Update 按 rowid 更新记录。
-	Update(rowid any, item reflux.R) error
-
-	// Delete 按 rowid 删除记录。
-	Delete(rowid any) error
-}
-
-type BaseResVirtualTable struct {
-	Controller
-
-	res   req.Resource
-	cache *db.Database
-}
-
-func (b *BaseResVirtualTable) Cache() *db.Database {
-	// CREATE TABLE IF NOT EXISTS ` + res.Table + ` (
-	// 	id         INTEGER PRIMARY KEY AUTOINCREMENT,
-	// 	kind       VARCHAR(8)   NOT NULL DEFAULT 'list',
-	// 	key        VARCHAR(255) NOT NULL,
-	// 	lmt        INTEGER      NOT NULL DEFAULT 0,
-	// 	page       INTEGER      NOT NULL DEFAULT 0,
-	// 	data       TEXT         NOT NULL,
-	// 	total      INTEGER      NOT NULL DEFAULT 0,
-	// 	expires_at BIGINT       NOT NULL,
-	// 	UNIQUE(kind, key, lmt, page)
-	// )
-	return b.cache
-}
-
-func (b *BaseResVirtualTable) GetResource() req.Resource {
-	return b.res
-}
-
-func (b *BaseResVirtualTable) GetExtraFields() []req.ResField {
-	return nil
-}
-
-func (b *BaseResVirtualTable) Fetch(pk any) (map[string]any, error) {
-	return nil, nil
-}
-
-func (b *BaseResVirtualTable) Detail(detail reflux.R) (changed bool, err error) {
-	return false, nil
-}
-
-func (b *BaseResVirtualTable) Select(filter reflux.R, pagesize, page int) (items []any, total int64, err error) {
-	return nil, 0, ErrFuncNotImplemented
-}
-
-func (b *BaseResVirtualTable) Insert(item reflux.R) (rowid int64, err error) {
-	return 0, ErrFuncNotImplemented
-}
-
-func (b *BaseResVirtualTable) Update(rowid any, item reflux.R) error {
-	return ErrFuncNotImplemented
-}
-
-func (b *BaseResVirtualTable) Delete(rowid any) error {
-	return ErrFuncNotImplemented
-}
-
-func NewBaseResVirtualTable(container do.Injector, res req.Resource, conn *db.Database) ResVirtualTable {
-	return &BaseResVirtualTable{
-		Controller: &BaseController{container},
-		res:        res,
-		cache:      conn,
-	}
 }

support/service.go

@@ -21,10 +21,10 @@ func GoPluginCore(core string, services ...any) {
 			} else if fn, ok := f.(func(map[string]contracts.Controller) error); ok {
 				fn(s)
 			}
-		case map[string]contracts.ResListener:
+		case map[string]contracts.ResWatcher:
-			if f, err := p.Lookup("RegListens"); err != nil {
+			if f, err := p.Lookup("RegResWatchers"); err != nil {
 				panic(err)
-			} else if fn, ok := f.(func(map[string]contracts.ResListener) error); ok {
+			} else if fn, ok := f.(func(map[string]contracts.ResWatcher) error); ok {
 				fn(s)
 			}
 		case map[string]contracts.GRpc:

ws.go

@@ -5,17 +5,12 @@ import (
 
 	"git.fsdpf.net/go/req"
 	"github.com/gorilla/websocket"
-	"github.com/samber/do/v2"
 )
 
 type WsClientGroup string
 type WsClientID uint
 
-// type WsClient struct {
+const DefaultWsClientGroup WsClientGroup = "__DEFAULT__"
-// 	User
-// 	Group  WsClientGroup
-// 	Socket *websocket.Conn
-// }
 
 type WsClient interface {
 	req.User
@@ -50,50 +45,7 @@ type WsController interface {
 	// 获取 Ws 客户端
 	Client() WsClient
 	// 获取 Ws 标识
-	ClientId(req.GlobalParams) WsClientID
+	ClientID(req.GlobalParams) WsClientID
 	// 获取 Ws 分组
 	ClientGroup(req.GlobalParams) WsClientGroup
 }
-
-type WsHandleController struct {
-	Controller
-	ws      WsClient
-	request *http.Request
-}
-
-func (this WsHandleController) Request() *http.Request {
-	return this.request
-}
-
-func (this WsHandleController) Route() req.Route {
-	return this.Request().Context().Value(req.RouteCtx{Name: "Route"}).(req.Route)
-}
-
-func (this WsHandleController) User() req.User {
-	return this.Request().Context().Value(req.RouteCtx{Name: "User"}).(req.User)
-}
-
-func (this WsHandleController) Execute(req.GlobalParams) error {
-	return ErrFuncNotImplemented
-}
-
-func (this WsHandleController) Client() WsClient {
-	return this.ws
-}
-
-func (WsHandleController) ClientId(req.GlobalParams) WsClientID {
-	wsClientID++
-	return wsClientID
-}
-
-func (WsHandleController) ClientGroup(req.GlobalParams) WsClientGroup {
-	return defaultWsClientGroup
-}
-
-func NewWsController(container do.Injector, request *http.Request, ws WsClient) WsController {
-	return &WsHandleController{
-		Controller: &BaseController{container},
-		request:    request,
-		ws:         ws,
-	}
-}