diff --git a/app.go b/app.go index 070180d..6522f4a 100644 --- a/app.go +++ b/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) diff --git a/condflow.go b/condflow.go index e024cda..f8631ff 100644 --- a/condflow.go +++ b/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, - } -} diff --git a/controller.go b/controller.go index 2672e13..b72c32f 100644 --- a/controller.go +++ b/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} -} diff --git a/cron.go b/cron.go index 2a30b5a..e256e7a 100644 --- a/cron.go +++ b/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 -} diff --git a/event_stream.go b/event_stream.go index 4aec3bb..39a478f 100644 --- a/event_stream.go +++ b/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), - } -} diff --git a/executor.go b/executor.go index 258b86f..7a0524c 100644 --- a/executor.go +++ b/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, - } -} diff --git a/gobridge.go b/gobridge.go index ea75307..7fc14a3 100644 --- a/gobridge.go +++ b/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() {} diff --git a/grpc.go b/grpc.go index 6803920..d6b7f01 100644 --- a/grpc.go +++ b/grpc.go @@ -16,10 +16,3 @@ type GRpc interface { GetGRpcServiceDesc() *grpc.ServiceDesc } -type GRpcBase struct { - Controller -} - -func (GRpcBase) GetGRpcServiceDesc() *grpc.ServiceDesc { - return nil -} diff --git a/job.go b/job.go index 27c77e8..6c44e7e 100644 --- a/job.go +++ b/job.go @@ -13,12 +13,18 @@ type Job interface { Controller // 任务处理 Handle(any, req.User) error + // OnExceeded 重试次数耗尽时由框架回调,用于清理状态。默认实现为空操作。 + OnExceeded(any, req.User) error } -type JobBase struct { - Controller +// NoRetryError 表示不可重试的错误,job 返回此类型后框架立即标记为 exceeded,不再调度重试。 +type NoRetryError struct { + Err error } -func (JobBase) Handle(any, req.User) error { - return nil -} +func (e *NoRetryError) Error() string { return e.Err.Error() } +func (e *NoRetryError) Unwrap() error { return e.Err } + +// ErrNoRetry 将 err 包装为不可重试错误,job handler 使用此函数终止重试。 +func ErrNoRetry(err error) error { return &NoRetryError{Err: err} } + diff --git a/mcp.go b/mcp.go index 46bf364..aab3dc9 100644 --- a/mcp.go +++ b/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 } diff --git a/mqtt.go b/mqtt.go index 9b575db..d89bfa4 100644 --- a/mqtt.go +++ b/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}, - } -} diff --git a/res_virtual_table.go b/res_virtual_table.go index c6b4544..7ddf9d3 100644 --- a/res_virtual_table.go +++ b/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 机制调用。 -// Select 提供列表查询,Fetch 提供单条完整查询(含 Detail 补充字段), -// Insert/Update/Delete 支持写操作并自动同步缓存。 +// ResVirtualTableSetup 由 framework-v2 在 sqlite_vtable build tag 下注册到容器,负责向 SQLite +// 注册虚拟表模块,具体说明见 req.ResVirtualTableSetup。 +type ResVirtualTableSetup = req.ResVirtualTableSetup + +// ResVirtualTable 在 req.ResVirtualTable 基础上合并 Controller,使实现方可以访问 DI 容器等框架能力; +// 具体方法说明见 req.ResVirtualTable。 type ResVirtualTable interface { Controller - GetResource() req.Resource - - // 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, - } + req.ResVirtualTable } diff --git a/support/service.go b/support/service.go index 47e5c5c..51637e2 100644 --- a/support/service.go +++ b/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: - if f, err := p.Lookup("RegListens"); err != nil { + case map[string]contracts.ResWatcher: + 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: diff --git a/ws.go b/ws.go index afc41a8..ea0fa85 100644 --- a/ws.go +++ b/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 { -// User -// Group WsClientGroup -// Socket *websocket.Conn -// } +const DefaultWsClientGroup WsClientGroup = "__DEFAULT__" 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, - } -}