mirror of
https://github.com/Tencent/WeKnora.git
synced 2026-10-03 22:38:42 +08:00
feat(plugin): Host API with a key-value store, and document parsers from plugins (#3721)
* feat(plugin): Host API with plugin tokens and a key-value store * feat(plugin): document parsers from code plugins * docs(pluginsdk): parsers and the Host API
This commit is contained in:
@@ -0,0 +1,130 @@
|
||||
// Command subtitles is an example parser plugin: it turns SRT and WebVTT
|
||||
// subtitle files into a Markdown transcript, one paragraph per cue with its
|
||||
// start time. It also keeps a per-workspace count of parsed files through
|
||||
// the Host API key-value store, to show how a plugin keeps state without a
|
||||
// database.
|
||||
//
|
||||
// Build a package with ./package.sh.
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Tencent/WeKnora/pluginsdk"
|
||||
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
|
||||
)
|
||||
|
||||
// Version must match plugin.yaml.
|
||||
const Version = "1.0.0"
|
||||
|
||||
func main() {
|
||||
p := pluginsdk.New(pluginsdk.Info{ID: "weknora-examples.subtitles", Version: Version})
|
||||
p.Parser("subtitles", pluginsdk.ParserFunc(parse))
|
||||
if err := p.Serve(); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// timing matches a cue timing line: 00:01:02,500 --> 00:01:04,000 (SRT) or
|
||||
// 01:02.500 --> 01:04.000 (WebVTT, hours optional).
|
||||
var timing = regexp.MustCompile(`^\s*((?:\d+:)?\d{2}:\d{2}[.,]\d{3})\s*-->\s*`)
|
||||
|
||||
// tags strips WebVTT / SRT inline markup such as <i> or <v Speaker>.
|
||||
var tags = regexp.MustCompile(`<[^>]*>`)
|
||||
|
||||
type cue struct {
|
||||
start string
|
||||
text []string
|
||||
}
|
||||
|
||||
func parse(ctx context.Context, call *pluginsdk.Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
|
||||
if len(in.Content) == 0 {
|
||||
return nil, pluginapi.Errorf(pluginapi.CodeInvalidConfig, "the subtitle file is empty")
|
||||
}
|
||||
cues := parseCues(in.Content)
|
||||
if len(cues) == 0 {
|
||||
return nil, pluginapi.Errorf(
|
||||
pluginapi.CodeInvalidConfig,
|
||||
"no subtitle cues found; is this an .srt or .vtt file?",
|
||||
)
|
||||
}
|
||||
title := strings.TrimSpace(in.Title)
|
||||
if title == "" {
|
||||
title = strings.TrimSuffix(in.FileName, "."+in.FileType)
|
||||
}
|
||||
var md strings.Builder
|
||||
fmt.Fprintf(&md, "# %s\n\n", title)
|
||||
for _, c := range cues {
|
||||
fmt.Fprintf(&md, "**[%s]** %s\n\n", shortTime(c.start), strings.Join(c.text, " "))
|
||||
}
|
||||
countParsed(ctx, call)
|
||||
return &pluginapi.ParseOutput{
|
||||
Markdown: strings.TrimSpace(md.String()) + "\n",
|
||||
Metadata: map[string]string{"cues": fmt.Sprint(len(cues))},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func parseCues(data []byte) []cue {
|
||||
data = bytes.TrimPrefix(data, []byte("\xef\xbb\xbf"))
|
||||
var out []cue
|
||||
var cur *cue
|
||||
sc := bufio.NewScanner(bytes.NewReader(data))
|
||||
sc.Buffer(make([]byte, 0, 64<<10), 1<<20)
|
||||
for sc.Scan() {
|
||||
line := strings.TrimRight(sc.Text(), "\r")
|
||||
if m := timing.FindStringSubmatch(line); m != nil {
|
||||
out = append(out, cue{start: m[1]})
|
||||
cur = &out[len(out)-1]
|
||||
continue
|
||||
}
|
||||
if strings.TrimSpace(line) == "" {
|
||||
cur = nil
|
||||
continue
|
||||
}
|
||||
if cur == nil {
|
||||
continue // cue numbers, WEBVTT header, NOTE blocks
|
||||
}
|
||||
if text := strings.TrimSpace(tags.ReplaceAllString(line, "")); text != "" {
|
||||
cur.text = append(cur.text, text)
|
||||
}
|
||||
}
|
||||
kept := out[:0]
|
||||
for _, c := range out {
|
||||
if len(c.text) > 0 {
|
||||
kept = append(kept, c)
|
||||
}
|
||||
}
|
||||
return kept
|
||||
}
|
||||
|
||||
// shortTime drops milliseconds and a zero hour: 00:01:02,500 → 01:02.
|
||||
func shortTime(t string) string {
|
||||
t = strings.NewReplacer(",", ".").Replace(t)
|
||||
if i := strings.IndexByte(t, '.'); i >= 0 {
|
||||
t = t[:i]
|
||||
}
|
||||
return strings.TrimPrefix(t, "00:")
|
||||
}
|
||||
|
||||
// countParsed bumps the workspace's parsed-file counter. It is best effort:
|
||||
// a missing grant or an unreachable host must not fail the parse.
|
||||
func countParsed(ctx context.Context, call *pluginsdk.Call) {
|
||||
host := call.Host()
|
||||
if host == nil {
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
var n int
|
||||
if _, err := host.KVGet(ctx, "stats/parsed", &n); err != nil {
|
||||
return
|
||||
}
|
||||
_ = host.KVPut(ctx, "stats/parsed", n+1, 0)
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"go/parser"
|
||||
"go/token"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/google/uuid"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/application/repository"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/hostapi"
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
"github.com/Tencent/WeKnora/pluginsdk"
|
||||
"github.com/Tencent/WeKnora/pluginsdk/client"
|
||||
"github.com/Tencent/WeKnora/pluginsdk/conformance"
|
||||
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
|
||||
)
|
||||
|
||||
const srt = "\ufeff1\r\n00:00:01,000 --> 00:00:03,500\r\nHello <i>world</i>.\r\n\r\n" +
|
||||
"2\r\n00:01:02,250 --> 00:01:04,000\r\nSecond line\r\ncontinues here.\r\n"
|
||||
|
||||
const vtt = `WEBVTT
|
||||
|
||||
NOTE a comment
|
||||
|
||||
01:02.500 --> 01:04.000
|
||||
<v Ada>Plugins run out of process.
|
||||
|
||||
1:00:00.000 --> 1:00:02.000 align:start
|
||||
Bye.
|
||||
`
|
||||
|
||||
func TestParseSRTAndVTT(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
out, err := parse(
|
||||
ctx,
|
||||
&pluginsdk.Call{},
|
||||
pluginapi.ParseInput{FileName: "talk.srt", FileType: "srt", Content: []byte(srt)},
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := "# talk\n\n**[00:01]** Hello world.\n\n**[01:02]** Second line continues here.\n"
|
||||
if out.Markdown != want || out.Metadata["cues"] != "2" {
|
||||
t.Fatalf("srt markdown = %q", out.Markdown)
|
||||
}
|
||||
out, err = parse(
|
||||
ctx,
|
||||
&pluginsdk.Call{},
|
||||
pluginapi.ParseInput{FileName: "t.vtt", FileType: "vtt", Title: "Talk", Content: []byte(vtt)},
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.Contains(out.Markdown, "# Talk") ||
|
||||
!strings.Contains(out.Markdown, "**[01:02]** Plugins run out of process.") ||
|
||||
!strings.Contains(out.Markdown, "**[1:00:00]** Bye.") {
|
||||
t.Fatalf("vtt markdown = %q", out.Markdown)
|
||||
}
|
||||
for _, bad := range []string{"", "just some text\nwith no cues\n"} {
|
||||
_, err := parse(
|
||||
ctx,
|
||||
&pluginsdk.Call{},
|
||||
pluginapi.ParseInput{FileName: "x.srt", FileType: "srt", Content: []byte(bad)},
|
||||
)
|
||||
if e, ok := pluginapi.AsError(err); !ok || e.Code != pluginapi.CodeInvalidConfig || e.Retryable {
|
||||
t.Errorf("%q: want a permanent invalid_config error, got %v", bad, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseCountsThroughTheHostAPI(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.AutoMigrate(&types.PluginKV{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
iss := hostapi.NewIssuer([]byte("k"))
|
||||
r := gin.New()
|
||||
hostapi.NewHandler(iss, hostapi.NewKV(repository.NewPluginKVRepository(db))).Register(r)
|
||||
srv := httptest.NewServer(r)
|
||||
defer srv.Close()
|
||||
|
||||
tok, _, _ := iss.Issue("weknora-examples.subtitles", Version, 3, []string{"kv"})
|
||||
call := &pluginsdk.Call{
|
||||
Context: pluginapi.Context{TenantID: 3, Host: &pluginapi.HostAccess{URL: srv.URL, Token: tok}},
|
||||
}
|
||||
in := pluginapi.ParseInput{FileName: "a.srt", FileType: "srt", Content: []byte(srt)}
|
||||
for range 2 {
|
||||
if _, err := parse(context.Background(), call, in); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
var n int
|
||||
if ok, err := call.Host().KVGet(context.Background(), "stats/parsed", &n); !ok || err != nil || n != 2 {
|
||||
t.Fatalf("parsed count = %d (%v, %v)", n, ok, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConformsAndStaysExternal(t *testing.T) {
|
||||
p := pluginsdk.New(pluginsdk.Info{ID: "weknora-examples.subtitles", Version: Version})
|
||||
p.Parser("subtitles", pluginsdk.ParserFunc(parse))
|
||||
srv := httptest.NewServer(p.Handler())
|
||||
defer srv.Close()
|
||||
rep := conformance.Run(context.Background(), conformance.Target{
|
||||
Client: client.New(srv.URL, nil, nil),
|
||||
Raw: func(ctx context.Context, path string, body []byte) (*http.Response, error) {
|
||||
req, _ := http.NewRequestWithContext(ctx, http.MethodPost, srv.URL+path, bytes.NewReader(body))
|
||||
return http.DefaultClient.Do(req)
|
||||
},
|
||||
})
|
||||
for _, res := range rep.Results {
|
||||
if !res.Passed {
|
||||
t.Errorf("%s: %s", res.Name, res.Detail)
|
||||
}
|
||||
}
|
||||
files, _ := filepath.Glob("*.go")
|
||||
for _, f := range files {
|
||||
if strings.HasSuffix(f, "_test.go") {
|
||||
continue
|
||||
}
|
||||
src, _ := os.ReadFile(f)
|
||||
parsed, err := parser.ParseFile(token.NewFileSet(), f, src, parser.ImportsOnly)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, imp := range parsed.Imports {
|
||||
if strings.Contains(imp.Path.Value, "WeKnora/internal") {
|
||||
t.Errorf("%s imports %s", f, imp.Path.Value)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Executable
+20
@@ -0,0 +1,20 @@
|
||||
#!/usr/bin/env bash
|
||||
# Builds the subtitles example plugin for the platforms WeKnora runs on and zips it
|
||||
# into a .wkp package: plugin.yaml and bin/<os>-<arch>/subtitles.
|
||||
set -euo pipefail
|
||||
cd "$(dirname "$0")"
|
||||
version=$(sed -n 's/^version: //p' plugin.yaml)
|
||||
out=${1:-"weknora-examples-subtitles-${version}.wkp"}
|
||||
stage=$(mktemp -d)
|
||||
trap 'rm -rf "$stage"' EXIT
|
||||
cp plugin.yaml "$stage"/
|
||||
for target in linux/amd64 linux/arm64 darwin/amd64 darwin/arm64 windows/amd64; do
|
||||
os=${target%/*}
|
||||
arch=${target#*/}
|
||||
bin="$stage/bin/${os}-${arch}/subtitles"
|
||||
[ "$os" = windows ] && bin="$bin.exe"
|
||||
CGO_ENABLED=0 GOOS=$os GOARCH=$arch go build -trimpath -ldflags="-s -w" -o "$bin" .
|
||||
done
|
||||
rm -f "$out"
|
||||
(cd "$stage" && zip -qr - .) > "$out"
|
||||
echo "wrote $out"
|
||||
@@ -0,0 +1,23 @@
|
||||
schemaVersion: 1
|
||||
id: weknora-examples.subtitles
|
||||
version: 1.0.0
|
||||
apiVersion: weknora.plugin/v1
|
||||
name: { en-US: Subtitles, zh-CN: 字幕解析 }
|
||||
description:
|
||||
en-US: Parse SRT and WebVTT subtitles into a searchable transcript.
|
||||
zh-CN: 把 SRT、WebVTT 字幕解析为可检索的文字稿。
|
||||
publisher: { id: weknora-examples, name: WeKnora examples }
|
||||
homepage: https://github.com/Tencent/WeKnora/tree/main/examples/plugins/subtitles
|
||||
license: MIT
|
||||
runtime: { type: host, kind: binary, entry: "bin/{os}-{arch}/subtitles" }
|
||||
permissions:
|
||||
# Counts parsed files per workspace in the Host API key-value store.
|
||||
hostApi: [kv]
|
||||
contributes:
|
||||
parsers:
|
||||
- id: subtitles
|
||||
name: { en-US: Subtitles, zh-CN: 字幕解析 }
|
||||
description:
|
||||
en-US: SRT and WebVTT subtitles as a transcript with cue times
|
||||
zh-CN: 带时间点的字幕文字稿
|
||||
fileTypes: [srt, vtt]
|
||||
@@ -107,6 +107,10 @@ export interface ParserEngineInfo {
|
||||
FileTypes: string[]
|
||||
Available?: boolean
|
||||
UnavailableReason?: string
|
||||
/** Set on engines an installed plugin provides (Name is then acme.ocr/ocr). */
|
||||
PluginID?: string
|
||||
/** Localized names of plugin engines, keyed by locale; "default" as fallback. */
|
||||
DisplayNames?: Record<string, string>
|
||||
}
|
||||
|
||||
/** 解析引擎配置(引擎连接参数存空间;聊天附件解析策略在智能体中配置) */
|
||||
|
||||
@@ -66,6 +66,7 @@
|
||||
</template>
|
||||
|
||||
<script setup lang="ts">
|
||||
import { pickLocale } from '@/utils/localizedText'
|
||||
import { ref, watch, computed, onMounted, onUnmounted } from 'vue'
|
||||
import { useI18n } from 'vue-i18n'
|
||||
import { type ParserEngineInfo } from '@/api/system'
|
||||
@@ -73,10 +74,13 @@ import { useEditorResourcesStore } from '@/stores/editorResources'
|
||||
import { useUIStore } from '@/stores/ui'
|
||||
import { storeToRefs } from 'pinia'
|
||||
|
||||
const { t } = useI18n()
|
||||
const { t, locale } = useI18n()
|
||||
const editorResources = useEditorResourcesStore()
|
||||
|
||||
function getEngineDisplayName(engineName: string): string {
|
||||
// Plugin engines name themselves.
|
||||
const names = parserEngines.value.find((e) => e.Name === engineName)?.DisplayNames
|
||||
if (names) return pickLocale(names, locale.value) || names.default || engineName
|
||||
const key = `kbSettings.parser.engines.${engineName}.name`
|
||||
const translated = t(key)
|
||||
return translated !== key ? translated : engineName
|
||||
|
||||
@@ -409,6 +409,7 @@
|
||||
</template>
|
||||
|
||||
<script setup lang="ts">
|
||||
import { pickLocale } from '@/utils/localizedText'
|
||||
import { ref, computed, onMounted, nextTick } from 'vue'
|
||||
import { useI18n } from 'vue-i18n'
|
||||
import { useUIStore } from '@/stores/ui'
|
||||
@@ -425,7 +426,7 @@ import {
|
||||
} from '@/api/system'
|
||||
import { getWeKnoraCloudStatus } from '@/api/model'
|
||||
|
||||
const { t } = useI18n()
|
||||
const { t, locale } = useI18n()
|
||||
const uiStore = useUIStore()
|
||||
const authStore = useAuthStore()
|
||||
|
||||
@@ -546,6 +547,9 @@ function engineInitial(engineName: string): string {
|
||||
}
|
||||
|
||||
function getEngineDisplayName(engineName: string): string {
|
||||
// Plugin engines name themselves.
|
||||
const names = engines.value.find((e) => e.Name === engineName)?.DisplayNames
|
||||
if (names) return pickLocale(names, locale.value) || names.default || engineName
|
||||
const key = `kbSettings.parser.engines.${engineName}.name`
|
||||
const translated = t(key)
|
||||
return translated !== key ? translated : engineName
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
"github.com/Tencent/WeKnora/internal/types/interfaces"
|
||||
)
|
||||
|
||||
// pluginKVRepository stores plugins' Host API key-value data (plugin_kv,
|
||||
// migration 000117).
|
||||
type pluginKVRepository struct {
|
||||
db *gorm.DB
|
||||
}
|
||||
|
||||
// NewPluginKVRepository wires the repository into the container.
|
||||
func NewPluginKVRepository(db *gorm.DB) interfaces.PluginKVRepository {
|
||||
return &pluginKVRepository{db: db}
|
||||
}
|
||||
|
||||
// live restricts a query to entries that have not expired.
|
||||
func live(q *gorm.DB, now time.Time) *gorm.DB {
|
||||
return q.Where("expires_at IS NULL OR expires_at > ?", now)
|
||||
}
|
||||
|
||||
func (r *pluginKVRepository) scope(ctx context.Context, pluginID string, tenantID uint64) *gorm.DB {
|
||||
return r.db.WithContext(ctx).Model(&types.PluginKV{}).
|
||||
Where("plugin_id = ? AND tenant_id = ?", pluginID, tenantID)
|
||||
}
|
||||
|
||||
func (r *pluginKVRepository) Get(
|
||||
ctx context.Context, pluginID string, tenantID uint64, key string,
|
||||
) (*types.PluginKV, error) {
|
||||
var rows []types.PluginKV
|
||||
err := live(r.scope(ctx, pluginID, tenantID).Where("key = ?", key), time.Now()).Limit(1).Find(&rows).Error
|
||||
if err != nil || len(rows) == 0 {
|
||||
return nil, err
|
||||
}
|
||||
return &rows[0], nil
|
||||
}
|
||||
|
||||
func (r *pluginKVRepository) Put(ctx context.Context, e *types.PluginKV) error {
|
||||
return r.db.WithContext(ctx).Clauses(clause.OnConflict{
|
||||
Columns: []clause.Column{{Name: "plugin_id"}, {Name: "tenant_id"}, {Name: "key"}},
|
||||
DoUpdates: clause.AssignmentColumns([]string{"value", "expires_at", "updated_at"}),
|
||||
}).Create(e).Error
|
||||
}
|
||||
|
||||
func (r *pluginKVRepository) Delete(ctx context.Context, pluginID string, tenantID uint64, key string) (bool, error) {
|
||||
res := r.db.WithContext(ctx).
|
||||
Where("plugin_id = ? AND tenant_id = ? AND key = ?", pluginID, tenantID, key).
|
||||
Delete(&types.PluginKV{})
|
||||
return res.RowsAffected > 0, res.Error
|
||||
}
|
||||
|
||||
// escapeLike makes a prefix match literally in LIKE.
|
||||
func escapeLike(s string) string {
|
||||
return strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(s)
|
||||
}
|
||||
|
||||
func (r *pluginKVRepository) List(
|
||||
ctx context.Context, pluginID string, tenantID uint64, prefix, after string, limit int,
|
||||
) ([]types.PluginKV, error) {
|
||||
q := live(r.scope(ctx, pluginID, tenantID), time.Now())
|
||||
if prefix != "" {
|
||||
q = q.Where(`key LIKE ? ESCAPE '\'`, escapeLike(prefix)+"%")
|
||||
}
|
||||
if after != "" {
|
||||
q = q.Where("key > ?", after)
|
||||
}
|
||||
var rows []types.PluginKV
|
||||
err := q.Order("key ASC").Limit(limit).Find(&rows).Error
|
||||
return rows, err
|
||||
}
|
||||
|
||||
func (r *pluginKVRepository) Count(ctx context.Context, pluginID string, tenantID uint64) (int64, error) {
|
||||
var n int64
|
||||
err := live(r.scope(ctx, pluginID, tenantID), time.Now()).Count(&n).Error
|
||||
return n, err
|
||||
}
|
||||
|
||||
func (r *pluginKVRepository) DeleteExpired(ctx context.Context, now time.Time) (int64, error) {
|
||||
res := r.db.WithContext(ctx).Where("expires_at IS NOT NULL AND expires_at <= ?", now).Delete(&types.PluginKV{})
|
||||
return res.RowsAffected, res.Error
|
||||
}
|
||||
@@ -93,3 +93,52 @@ func TestPluginTenantSettingUpsertColumns(t *testing.T) {
|
||||
require.False(t, got.Enabled)
|
||||
require.JSONEq(t, `{"region":"eu"}`, string(got.Config), "flipping the switch must keep the config")
|
||||
}
|
||||
|
||||
func TestPluginKVRepository(t *testing.T) {
|
||||
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, db.AutoMigrate(&types.PluginKV{}))
|
||||
repo := NewPluginKVRepository(db)
|
||||
ctx := context.Background()
|
||||
put := func(tenant uint64, key, value string, expires *time.Time) {
|
||||
t.Helper()
|
||||
require.NoError(t, repo.Put(ctx, &types.PluginKV{
|
||||
PluginID: "acme.x", TenantID: tenant, Key: key, Value: types.JSON(value),
|
||||
ExpiresAt: expires, UpdatedAt: time.Now(),
|
||||
}))
|
||||
}
|
||||
past := time.Now().Add(-time.Minute)
|
||||
put(1, "cursor:a", `1`, nil)
|
||||
put(1, "cursor:b", `2`, nil)
|
||||
put(1, "cursor_x", `3`, nil) // "_" must not act as a LIKE wildcard
|
||||
put(1, "old", `4`, &past)
|
||||
put(2, "cursor:a", `99`, nil)
|
||||
put(1, "cursor:a", `10`, nil) // upsert
|
||||
|
||||
got, err := repo.Get(ctx, "acme.x", 1, "cursor:a")
|
||||
require.NoError(t, err)
|
||||
require.JSONEq(t, `10`, string(got.Value))
|
||||
got, err = repo.Get(ctx, "acme.x", 1, "old")
|
||||
require.NoError(t, err)
|
||||
require.Nil(t, got, "expired entries read as missing")
|
||||
|
||||
list, err := repo.List(ctx, "acme.x", 1, "cursor:", "", 10)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, list, 2)
|
||||
list, err = repo.List(ctx, "acme.x", 1, "cursor:", "cursor:a", 10)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, list, 1)
|
||||
require.Equal(t, "cursor:b", list[0].Key)
|
||||
|
||||
n, err := repo.Count(ctx, "acme.x", 1)
|
||||
require.NoError(t, err)
|
||||
require.EqualValues(t, 3, n)
|
||||
removed, err := repo.DeleteExpired(ctx, time.Now())
|
||||
require.NoError(t, err)
|
||||
require.EqualValues(t, 1, removed)
|
||||
ok, err := repo.Delete(ctx, "acme.x", 1, "cursor:b")
|
||||
require.NoError(t, err)
|
||||
require.True(t, ok)
|
||||
got, _ = repo.Get(ctx, "acme.x", 2, "cursor:a")
|
||||
require.JSONEq(t, `99`, string(got.Value), "tenants are separate partitions")
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"mime/multipart"
|
||||
"net/http"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -56,8 +57,11 @@ func isSupportedImportExtension(ext string) bool {
|
||||
if ext == "" || ext == unknownFileType {
|
||||
return false
|
||||
}
|
||||
_, ok := supportedImportFileExtensions[ext]
|
||||
return ok
|
||||
if _, ok := supportedImportFileExtensions[ext]; ok {
|
||||
return true
|
||||
}
|
||||
// Installed plugins' parser engines bring their own file types.
|
||||
return slices.Contains(docparser.PluginFileTypes(), ext)
|
||||
}
|
||||
|
||||
// isValidFileType checks if a filename's extension is supported for import.
|
||||
|
||||
@@ -169,8 +169,11 @@ func BuildContainer(container *dig.Container) *dig.Container {
|
||||
must(container.Provide(activate.NewModelVendors))
|
||||
must(container.Provide(pluginhost.NewManager))
|
||||
must(container.Provide(newPluginInvoker))
|
||||
must(container.Provide(repository.NewPluginKVRepository))
|
||||
must(container.Provide(newPluginHostAPI))
|
||||
must(container.Provide(activate.NewWebSearch))
|
||||
must(container.Provide(activate.NewConnectors))
|
||||
must(container.Provide(activate.NewParsers))
|
||||
must(container.Provide(newMCPServiceRepository))
|
||||
must(container.Provide(repository.NewMCPToolApprovalRepository))
|
||||
must(container.Provide(repository.NewMCPOAuthRepository))
|
||||
@@ -1854,8 +1857,10 @@ func installPluginGate(
|
||||
webSearch *handler.WebSearchProviderHandler,
|
||||
dataSources *handler.DataSourceHandler,
|
||||
imHandler *handler.IMHandler,
|
||||
system *handler.SystemHandler,
|
||||
) {
|
||||
for _, h := range []interface{ SetPluginGate(interfaces.PluginGate) }{models, webSearch, dataSources, imHandler} {
|
||||
gated := []interface{ SetPluginGate(interfaces.PluginGate) }{models, webSearch, dataSources, imHandler, system}
|
||||
for _, h := range gated {
|
||||
h.SetPluginGate(gate)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,10 @@ package container
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
"go.uber.org/dig"
|
||||
@@ -13,6 +17,7 @@ import (
|
||||
"github.com/Tencent/WeKnora/internal/handler"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/activate"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/host"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/hostapi"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/install"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
|
||||
pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry"
|
||||
@@ -38,6 +43,7 @@ type pluginActivators struct {
|
||||
Host *host.Manager
|
||||
WebSearch *activate.WebSearch
|
||||
Connectors *activate.Connectors
|
||||
Parsers *activate.Parsers
|
||||
MCP *activate.MCPServers
|
||||
Skills *activate.Skills
|
||||
Vendors *activate.ModelVendors
|
||||
@@ -47,7 +53,29 @@ type pluginActivators struct {
|
||||
// list orders the activators: the host first, so a code plugin's process is
|
||||
// running before anything routes calls to it.
|
||||
func (a pluginActivators) list() []reconcile.Activator {
|
||||
return []reconcile.Activator{a.Host, a.WebSearch, a.Connectors, a.Vendors, a.MCP, a.Skills}
|
||||
return []reconcile.Activator{a.Host, a.WebSearch, a.Connectors, a.Parsers, a.Vendors, a.MCP, a.Skills}
|
||||
}
|
||||
|
||||
// newPluginHostAPI serves the Host API and gives calls a way back to it: the
|
||||
// embedded host's plugins reach this node on loopback.
|
||||
func newPluginHostAPI(
|
||||
cfg *config.Config, iv *activate.Invoker, repo interfaces.PluginKVRepository, cleaner interfaces.ResourceCleaner,
|
||||
) *hostapi.Handler {
|
||||
issuer := hostapi.NewIssuerFromEnv()
|
||||
url := strings.TrimSpace(os.Getenv("WEKNORA_PLUGIN_HOST_API_URL"))
|
||||
if url == "" {
|
||||
port := 8080
|
||||
if cfg != nil && cfg.Server != nil && cfg.Server.Port > 0 {
|
||||
port = cfg.Server.Port
|
||||
}
|
||||
url = fmt.Sprintf("http://127.0.0.1:%d", port)
|
||||
}
|
||||
iv.SetHostAPI(issuer, url)
|
||||
kv := hostapi.NewKV(repo)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
kv.StartSweeper(ctx, 10*time.Minute)
|
||||
cleaner.RegisterWithName("PluginKVSweeper", func() error { cancel(); return nil })
|
||||
return hostapi.NewHandler(issuer, kv)
|
||||
}
|
||||
|
||||
// bindPluginActivators hands the activators what they need once the plugin
|
||||
|
||||
@@ -38,6 +38,7 @@ var versionedSQLiteTables = []string{
|
||||
"plugin_tenant_settings",
|
||||
"plugins",
|
||||
"plugin_versions",
|
||||
"plugin_kv",
|
||||
}
|
||||
|
||||
// versionedSQLiteColumns maps each existing table to the columns that the
|
||||
@@ -72,7 +73,7 @@ var versionedSQLiteColumns = map[string][]string{
|
||||
}, // 000028
|
||||
}
|
||||
|
||||
const expectedSQLiteMigrationVersion = 35
|
||||
const expectedSQLiteMigrationVersion = 36
|
||||
|
||||
func TestSQLiteMigrationsCreateVersionedSchema(t *testing.T) {
|
||||
repoRoot := sqliteRepoRoot(t)
|
||||
|
||||
@@ -24,6 +24,7 @@ import (
|
||||
"github.com/Tencent/WeKnora/internal/infrastructure/docparser"
|
||||
"github.com/Tencent/WeKnora/internal/logger"
|
||||
modellimiter "github.com/Tencent/WeKnora/internal/models/limiter"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/manifest"
|
||||
"github.com/Tencent/WeKnora/internal/runtime"
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
"github.com/Tencent/WeKnora/internal/types/interfaces"
|
||||
@@ -38,6 +39,7 @@ type runtimeKnowledgeCanceller interface {
|
||||
|
||||
// SystemHandler handles system-related requests
|
||||
type SystemHandler struct {
|
||||
pluginGated
|
||||
cfg *config.Config
|
||||
neo4jDriver neo4j.Driver
|
||||
documentReader interfaces.DocumentReader
|
||||
@@ -420,7 +422,18 @@ func (h *SystemHandler) ListParserEngines(c *gin.Context) {
|
||||
connected := reader != nil && reader.IsConnected()
|
||||
remoteEngines := h.fetchRemoteEngines(c.Request.Context(), reader, overrides)
|
||||
engines := docparser.ListAllEngines(connected, overrides, remoteEngines)
|
||||
c.JSON(200, gin.H{"code": 0, "msg": "success", "data": engines, "docreader_addr": docreaderAddr, "docreader_transport": docreaderTransport, "connected": connected})
|
||||
// Engines of plugins the workspace has turned off leave the list.
|
||||
enabled := h.pluginFilter(c)
|
||||
visible := engines[:0]
|
||||
for _, e := range engines {
|
||||
if enabled(manifest.PointParsers, e.Name) {
|
||||
visible = append(visible, e)
|
||||
}
|
||||
}
|
||||
c.JSON(200, gin.H{
|
||||
"code": 0, "msg": "success", "data": visible,
|
||||
"docreader_addr": docreaderAddr, "docreader_transport": docreaderTransport, "connected": connected,
|
||||
})
|
||||
}
|
||||
|
||||
// ReconnectDocReader reconnects the document converter to a new (or same) DocReader address.
|
||||
|
||||
@@ -3,6 +3,9 @@ package docparser
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
"github.com/Tencent/WeKnora/internal/types/interfaces"
|
||||
@@ -41,32 +44,102 @@ type ReaderDeps struct {
|
||||
WeKnoraCloudCredentials func(ctx context.Context) *types.WeKnoraCloudCredentials
|
||||
}
|
||||
|
||||
// enginesMu guards the registries: plugin engines come and go while the
|
||||
// server runs.
|
||||
var enginesMu sync.RWMutex
|
||||
|
||||
// localEngines holds all locally registered parser engines, in registration
|
||||
// order — which is also the order the engine list is shown in.
|
||||
var localEngines []EngineRegistration
|
||||
|
||||
// pluginEngines holds engines installed plugins provide, by qualified name
|
||||
// ("acme.ocr/ocr"). They are kept apart from localEngines so the builtin
|
||||
// plugin catalog never describes them as builtins.
|
||||
var pluginEngines = map[string]EngineRegistration{}
|
||||
|
||||
// PluginEngineInfo is what a plugin engine adds to the engine list.
|
||||
type PluginEngineInfo interface {
|
||||
PluginID() string
|
||||
// DisplayNames localizes the engine's name, keyed by locale.
|
||||
DisplayNames() map[string]string
|
||||
}
|
||||
|
||||
// RegisterEngine adds an engine to the local registry. Called from init().
|
||||
func RegisterEngine(e EngineRegistration) {
|
||||
enginesMu.Lock()
|
||||
defer enginesMu.Unlock()
|
||||
localEngines = append(localEngines, e)
|
||||
}
|
||||
|
||||
// Engines returns the locally registered engines in registration order.
|
||||
// RegisterPluginEngine adds or replaces a plugin's engine. It refuses a name
|
||||
// a builtin engine uses.
|
||||
func RegisterPluginEngine(e EngineRegistration) error {
|
||||
enginesMu.Lock()
|
||||
defer enginesMu.Unlock()
|
||||
for _, builtin := range localEngines {
|
||||
if builtin.Name() == e.Name() {
|
||||
return fmt.Errorf("parser engine %s already exists", e.Name())
|
||||
}
|
||||
}
|
||||
pluginEngines[e.Name()] = e
|
||||
return nil
|
||||
}
|
||||
|
||||
// UnregisterPluginEngine removes a plugin's engine.
|
||||
func UnregisterPluginEngine(name string) {
|
||||
enginesMu.Lock()
|
||||
defer enginesMu.Unlock()
|
||||
delete(pluginEngines, name)
|
||||
}
|
||||
|
||||
// Engines returns the builtin engines in registration order.
|
||||
func Engines() []EngineRegistration {
|
||||
enginesMu.RLock()
|
||||
defer enginesMu.RUnlock()
|
||||
out := make([]EngineRegistration, len(localEngines))
|
||||
copy(out, localEngines)
|
||||
return out
|
||||
}
|
||||
|
||||
// lookupEngine returns the locally registered engine with this name.
|
||||
// pluginEngineList returns the plugin engines sorted by name.
|
||||
func pluginEngineList() []EngineRegistration {
|
||||
enginesMu.RLock()
|
||||
defer enginesMu.RUnlock()
|
||||
out := make([]EngineRegistration, 0, len(pluginEngines))
|
||||
for _, e := range pluginEngines {
|
||||
out = append(out, e)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Name() < out[j].Name() })
|
||||
return out
|
||||
}
|
||||
|
||||
// PluginFileTypes returns the file types plugin engines parse, so uploads
|
||||
// of those types are accepted.
|
||||
func PluginFileTypes() []string {
|
||||
var out []string
|
||||
for _, e := range pluginEngineList() {
|
||||
out = append(out, e.FileTypes(true)...)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// lookupEngine returns the registered engine with this name.
|
||||
func lookupEngine(name string) (EngineRegistration, bool) {
|
||||
enginesMu.RLock()
|
||||
defer enginesMu.RUnlock()
|
||||
for _, engine := range localEngines {
|
||||
if engine.Name() == name {
|
||||
return engine, true
|
||||
}
|
||||
}
|
||||
return nil, false
|
||||
e, ok := pluginEngines[name]
|
||||
return e, ok
|
||||
}
|
||||
|
||||
// isPluginEngineName reports whether a name is a qualified plugin
|
||||
// contribution ID, which no builtin or docreader engine uses.
|
||||
func isPluginEngineName(name string) bool { return strings.Contains(name, "/") }
|
||||
|
||||
// NewReader builds the reader for an engine.
|
||||
//
|
||||
// An empty engine name means "no explicit choice": simple formats are handled
|
||||
@@ -79,6 +152,11 @@ func NewReader(
|
||||
if registration, ok := lookupEngine(engine); ok {
|
||||
return registration.NewReader(ctx, deps)
|
||||
}
|
||||
if isPluginEngineName(engine) {
|
||||
// Its plugin was uninstalled or disabled: the Python docreader does
|
||||
// not know it either, so say so instead of sending it there.
|
||||
return nil, errEngineUnavailable(engine, "its plugin is not installed or not running")
|
||||
}
|
||||
if engine == "" && !isURL && IsSimpleFormat(fileType) {
|
||||
return &SimpleFormatReader{}, nil
|
||||
}
|
||||
@@ -112,10 +190,11 @@ func ListAllEngines(
|
||||
remoteMap[re.Name] = re
|
||||
}
|
||||
|
||||
seen := make(map[string]bool, len(localEngines))
|
||||
result := make([]types.ParserEngineInfo, 0, len(localEngines)+len(remoteEngines))
|
||||
locals := Engines()
|
||||
seen := make(map[string]bool, len(locals))
|
||||
result := make([]types.ParserEngineInfo, 0, len(locals)+len(remoteEngines))
|
||||
|
||||
for _, e := range localEngines {
|
||||
for _, e := range locals {
|
||||
name := e.Name()
|
||||
seen[name] = true
|
||||
|
||||
@@ -148,6 +227,18 @@ func ListAllEngines(
|
||||
result = append(result, re)
|
||||
}
|
||||
|
||||
for _, e := range pluginEngineList() {
|
||||
available, reason := e.CheckAvailable(docreaderConnected, overrides)
|
||||
info := types.ParserEngineInfo{
|
||||
Name: e.Name(), Description: e.Description(), FileTypes: e.FileTypes(docreaderConnected),
|
||||
Available: available, UnavailableReason: reason,
|
||||
}
|
||||
if meta, ok := e.(PluginEngineInfo); ok {
|
||||
info.PluginID, info.DisplayNames = meta.PluginID(), meta.DisplayNames()
|
||||
}
|
||||
result = append(result, info)
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
|
||||
@@ -28,6 +28,22 @@ type Invoker struct {
|
||||
mu sync.RWMutex
|
||||
tenancy *tenancy.Service
|
||||
plugins interfaces.PluginRepository
|
||||
// tokens and hostURL give plugins granted Host API scopes a way back.
|
||||
tokens TokenIssuer
|
||||
hostURL string
|
||||
}
|
||||
|
||||
// TokenIssuer signs the Host API token of one call.
|
||||
type TokenIssuer interface {
|
||||
Issue(pluginID, version string, tenantID uint64, scopes []string) (string, time.Time, error)
|
||||
}
|
||||
|
||||
// SetHostAPI lets calls carry Host API access: where plugins reach WeKnora
|
||||
// and how their tokens are signed.
|
||||
func (iv *Invoker) SetHostAPI(tokens TokenIssuer, url string) {
|
||||
iv.mu.Lock()
|
||||
iv.tokens, iv.hostURL = tokens, url
|
||||
iv.mu.Unlock()
|
||||
}
|
||||
|
||||
// NewInvoker creates an Invoker; Bind completes it.
|
||||
@@ -77,8 +93,15 @@ func (iv *Invoker) Envelope(
|
||||
env.Context.Deadline = &d
|
||||
}
|
||||
iv.mu.RLock()
|
||||
t, plugins := iv.tenancy, iv.plugins
|
||||
t, plugins, tokens, hostURL := iv.tenancy, iv.plugins, iv.tokens, iv.hostURL
|
||||
iv.mu.RUnlock()
|
||||
if tokens != nil && hostURL != "" && env.Context.TenantID != 0 && len(m.Permissions.HostAPI) > 0 {
|
||||
token, _, err := tokens.Issue(m.ID, m.Version, env.Context.TenantID, m.Permissions.HostAPI)
|
||||
if err != nil {
|
||||
return env, err
|
||||
}
|
||||
env.Context.Host = &pluginapi.HostAccess{URL: hostURL, Token: token}
|
||||
}
|
||||
if plugins != nil && len(m.Config.SystemSchema) > 0 {
|
||||
sys, _, err := install.OpenSystemConfig(ctx, plugins, m)
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
package activate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/plugin/hostapi"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/manifest"
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
)
|
||||
|
||||
func TestEnvelopeCarriesHostAccessOnlyForGrantedPlugins(t *testing.T) {
|
||||
iss := hostapi.NewIssuer([]byte("k"))
|
||||
iv := NewInvoker(fakeClients{})
|
||||
iv.SetHostAPI(iss, "http://127.0.0.1:9")
|
||||
ctx := context.WithValue(context.Background(), types.TenantIDContextKey, uint64(7))
|
||||
|
||||
granted := &manifest.Manifest{
|
||||
ID: "acme.x", Version: "1.0.0", Permissions: manifest.Permissions{HostAPI: []string{"kv"}},
|
||||
}
|
||||
env, err := iv.Envelope(ctx, granted, nil)
|
||||
if err != nil || env.Context.Host == nil || env.Context.Host.URL != "http://127.0.0.1:9" {
|
||||
t.Fatalf("envelope = %+v, %v", env.Context, err)
|
||||
}
|
||||
claims, err := iss.Verify(env.Context.Host.Token)
|
||||
if err != nil || claims.PluginID != "acme.x" || claims.TenantID != 7 || !claims.Has("kv") {
|
||||
t.Fatalf("token claims = %+v, %v", claims, err)
|
||||
}
|
||||
|
||||
plain := &manifest.Manifest{ID: "acme.y", Version: "1.0.0"}
|
||||
if env, _ := iv.Envelope(ctx, plain, nil); env.Context.Host != nil {
|
||||
t.Fatal("a plugin without Host API scopes must get no token")
|
||||
}
|
||||
if env, _ := iv.Envelope(context.Background(), granted, nil); env.Context.Host != nil {
|
||||
t.Fatal("a call without a tenant must get no token")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,152 @@
|
||||
package activate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"path"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/infrastructure/docparser"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/manifest"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
"github.com/Tencent/WeKnora/internal/types/interfaces"
|
||||
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
|
||||
)
|
||||
|
||||
// parseTimeout bounds one plugin parse when the caller set no deadline.
|
||||
const parseTimeout = 10 * time.Minute
|
||||
|
||||
// Parsers registers code plugins' document parsers as parser engines under
|
||||
// their qualified IDs, next to the builtin engines and the docreader's.
|
||||
type Parsers struct {
|
||||
iv *Invoker
|
||||
|
||||
mu sync.Mutex
|
||||
registered map[string][]string
|
||||
}
|
||||
|
||||
// NewParsers creates the parser activator.
|
||||
func NewParsers(iv *Invoker) *Parsers {
|
||||
return &Parsers{iv: iv, registered: map[string][]string{}}
|
||||
}
|
||||
|
||||
// Name implements reconcile.Activator.
|
||||
func (a *Parsers) Name() string { return "parsers" }
|
||||
|
||||
// Activate implements reconcile.Activator: all of a plugin's parsers or none.
|
||||
func (a *Parsers) Activate(_ context.Context, l *reconcile.Loaded) error {
|
||||
var ids []string
|
||||
for _, c := range l.Manifest.Contributes[manifest.PointParsers] {
|
||||
e := &pluginEngine{
|
||||
iv: a.iv, m: l.Manifest, local: c.ID, name: manifest.QualifiedID(l.Manifest.ID, c.ID),
|
||||
description: c.Description.Default, names: displayNames(c.Name), fileTypes: c.FileTypes,
|
||||
}
|
||||
if err := docparser.RegisterPluginEngine(e); err != nil {
|
||||
for _, done := range ids {
|
||||
docparser.UnregisterPluginEngine(done)
|
||||
}
|
||||
return fmt.Errorf("parser %s: %w", c.ID, err)
|
||||
}
|
||||
ids = append(ids, e.name)
|
||||
}
|
||||
a.mu.Lock()
|
||||
a.registered[l.Manifest.ID] = ids
|
||||
a.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Deactivate implements reconcile.Activator.
|
||||
func (a *Parsers) Deactivate(_ context.Context, pluginID string) error {
|
||||
a.mu.Lock()
|
||||
ids := a.registered[pluginID]
|
||||
delete(a.registered, pluginID)
|
||||
a.mu.Unlock()
|
||||
for _, id := range ids {
|
||||
docparser.UnregisterPluginEngine(id)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func displayNames(t manifest.LocalizedText) map[string]string {
|
||||
out := map[string]string{}
|
||||
for k, v := range t.Locales {
|
||||
out[k] = v
|
||||
}
|
||||
if t.Default != "" {
|
||||
out["default"] = t.Default
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// pluginEngine is a plugin parser as a docparser engine.
|
||||
type pluginEngine struct {
|
||||
iv *Invoker
|
||||
m *manifest.Manifest
|
||||
local, name string
|
||||
description string
|
||||
names map[string]string
|
||||
fileTypes []string
|
||||
}
|
||||
|
||||
var (
|
||||
_ docparser.EngineRegistration = (*pluginEngine)(nil)
|
||||
_ docparser.PluginEngineInfo = (*pluginEngine)(nil)
|
||||
)
|
||||
|
||||
func (e *pluginEngine) Name() string { return e.name }
|
||||
func (e *pluginEngine) Description() string { return e.description }
|
||||
func (e *pluginEngine) FileTypes(bool) []string { return e.fileTypes }
|
||||
func (e *pluginEngine) PluginID() string { return e.m.ID }
|
||||
func (e *pluginEngine) DisplayNames() map[string]string { return e.names }
|
||||
func (e *pluginEngine) CheckAvailable(bool, map[string]string) (bool, string) {
|
||||
if _, err := e.iv.clients.Client(e.m.ID); err != nil {
|
||||
return false, err.Error()
|
||||
}
|
||||
return true, ""
|
||||
}
|
||||
|
||||
// NewReader implements docparser.EngineRegistration. deps.Overrides carry
|
||||
// other engines' credentials and are deliberately not passed on; the
|
||||
// plugin's own settings travel in its system and tenant configuration.
|
||||
func (e *pluginEngine) NewReader(context.Context, docparser.ReaderDeps) (interfaces.DocReader, error) {
|
||||
return &remoteParser{engine: e}, nil
|
||||
}
|
||||
|
||||
// remoteParser reads one document through the plugin.
|
||||
type remoteParser struct{ engine *pluginEngine }
|
||||
|
||||
// Read implements interfaces.DocReader. A retryable plugin failure (the
|
||||
// plugin is down, rate limited) is returned as an error so the task is
|
||||
// retried; any other failure is final and reported in the result.
|
||||
func (r *remoteParser) Read(ctx context.Context, req *types.ReadRequest) (*types.ReadResult, error) {
|
||||
ctx, cancel := withDefaultTimeout(ctx, parseTimeout)
|
||||
defer cancel()
|
||||
in := pluginapi.ParseInput{
|
||||
FileName: req.FileName, FileType: strings.ToLower(strings.TrimPrefix(req.FileType, ".")),
|
||||
Content: req.FileContent, URL: req.URL, Title: req.Title,
|
||||
}
|
||||
var out pluginapi.ParseOutput
|
||||
err := r.engine.iv.Call(ctx, r.engine.m, pluginapi.ParsePath(r.engine.local), nil, in, &out)
|
||||
if err != nil {
|
||||
var pe *pluginapi.Error
|
||||
if errors.As(err, &pe) && !pe.Retryable {
|
||||
return &types.ReadResult{Error: err.Error()}, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
res := &types.ReadResult{MarkdownContent: out.Markdown, Metadata: out.Metadata}
|
||||
for _, img := range out.Images {
|
||||
if img.OriginalRef == "" || len(img.Data) == 0 {
|
||||
continue
|
||||
}
|
||||
res.ImageRefs = append(res.ImageRefs, types.ImageRef{
|
||||
Filename: path.Base(img.OriginalRef), OriginalRef: img.OriginalRef,
|
||||
MimeType: img.MimeType, ImageData: img.Data,
|
||||
})
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
@@ -0,0 +1,120 @@
|
||||
package activate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http/httptest"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/infrastructure/docparser"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/pkg"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/plugintest"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
"github.com/Tencent/WeKnora/pluginsdk"
|
||||
"github.com/Tencent/WeKnora/pluginsdk/client"
|
||||
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
|
||||
)
|
||||
|
||||
const ocrManifest = `schemaVersion: 1
|
||||
id: acme.ocr
|
||||
version: 1.0.0
|
||||
apiVersion: weknora.plugin/v1
|
||||
name: { en-US: ACME OCR }
|
||||
publisher: { id: acme }
|
||||
runtime: { type: host, kind: binary, entry: bin/ocr }
|
||||
contributes:
|
||||
parsers:
|
||||
- id: ocr
|
||||
name: { en-US: ACME OCR, zh-CN: ACME 识别 }
|
||||
fileTypes: [tiff, pdf]
|
||||
`
|
||||
|
||||
func TestPluginParser(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
plugin := pluginsdk.New(pluginsdk.Info{ID: "acme.ocr", Version: "1.0.0"})
|
||||
plugin.Parser("ocr", pluginsdk.ParserFunc(
|
||||
func(_ context.Context, _ *pluginsdk.Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
|
||||
switch string(in.Content) {
|
||||
case "busy":
|
||||
return nil, pluginapi.Errorf(pluginapi.CodeRateLimited, "slow down")
|
||||
case "broken":
|
||||
return nil, pluginapi.Errorf(pluginapi.CodeInvalidConfig, "not a scanned document")
|
||||
}
|
||||
return &pluginapi.ParseOutput{
|
||||
Markdown: "# " + in.FileName + "\n\n",
|
||||
Images: []pluginapi.ParsedImage{
|
||||
{OriginalRef: "page-1.png", MimeType: "image/png", Data: []byte{1, 2}},
|
||||
},
|
||||
Metadata: map[string]string{"pages": "1"},
|
||||
}, nil
|
||||
}))
|
||||
srv := httptest.NewServer(plugin.Handler())
|
||||
defer srv.Close()
|
||||
|
||||
p, err := pkg.Open(plugintest.Zip(t, map[string]string{"plugin.yaml": ocrManifest, "bin/ocr": "x"}))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
a := NewParsers(NewInvoker(fakeClients{client.New(srv.URL, nil, nil)}))
|
||||
if err := a.Activate(ctx, &reconcile.Loaded{Manifest: p.Manifest, Package: p}); err != nil {
|
||||
t.Fatalf("Activate: %v", err)
|
||||
}
|
||||
defer func() { _ = a.Deactivate(ctx, "acme.ocr") }()
|
||||
|
||||
var listed *types.ParserEngineInfo
|
||||
for _, e := range docparser.ListAllEngines(false, nil, nil) {
|
||||
if e.Name == "acme.ocr/ocr" {
|
||||
listed = &e
|
||||
}
|
||||
}
|
||||
if listed == nil || !listed.Available || listed.PluginID != "acme.ocr" ||
|
||||
listed.DisplayNames["zh-CN"] != "ACME 识别" ||
|
||||
!slices.Equal(listed.FileTypes, []string{"tiff", "pdf"}) {
|
||||
t.Fatalf("listed = %+v", listed)
|
||||
}
|
||||
if !slices.Contains(docparser.PluginFileTypes(), "tiff") {
|
||||
t.Fatal("plugin file types must be importable")
|
||||
}
|
||||
|
||||
reader, err := docparser.NewReader(ctx, "acme.ocr/ocr", "tiff", false, docparser.ReaderDeps{
|
||||
Overrides: map[string]string{"mineru_api_key": "must-not-leak"},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
res, err := reader.Read(
|
||||
ctx,
|
||||
&types.ReadRequest{FileName: "scan.tiff", FileType: ".TIFF", FileContent: []byte("img")},
|
||||
)
|
||||
if err != nil || !strings.HasPrefix(res.MarkdownContent, "# scan.tiff") || res.Metadata["pages"] != "1" ||
|
||||
len(
|
||||
res.ImageRefs,
|
||||
) != 1 || res.ImageRefs[0].OriginalRef != "page-1.png" || res.ImageRefs[0].Filename != "page-1.png" {
|
||||
t.Fatalf("Read = %+v, %v", res, err)
|
||||
}
|
||||
|
||||
if _, err := reader.Read(ctx, &types.ReadRequest{FileContent: []byte("busy")}); err == nil {
|
||||
t.Fatal("a retryable plugin error must come back as an error, so the task retries")
|
||||
}
|
||||
res, err = reader.Read(ctx, &types.ReadRequest{FileContent: []byte("broken")})
|
||||
if err != nil || res.Error != "not a scanned document" {
|
||||
t.Fatalf("a permanent plugin error must fail the document: %+v, %v", res, err)
|
||||
}
|
||||
|
||||
if err := a.Deactivate(ctx, "acme.ocr"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := docparser.NewReader(ctx, "acme.ocr/ocr", "tiff", false, docparser.ReaderDeps{}); err == nil ||
|
||||
!strings.Contains(err.Error(), "not installed or not running") {
|
||||
t.Fatalf("a removed plugin engine must not fall through to the docreader: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPluginParserCannotShadowABuiltin(t *testing.T) {
|
||||
e := &pluginEngine{name: "simple"}
|
||||
if err := docparser.RegisterPluginEngine(e); err == nil {
|
||||
t.Fatal("a plugin engine must not take a builtin engine's name")
|
||||
}
|
||||
}
|
||||
@@ -167,8 +167,10 @@ func (p *process) childEnv(socket, token string) []string {
|
||||
"HTTPS_PROXY=" + p.proxy.URL(),
|
||||
"http_proxy=" + p.proxy.URL(),
|
||||
"https_proxy=" + p.proxy.URL(),
|
||||
"NO_PROXY=",
|
||||
"no_proxy=",
|
||||
// The Host API is on this node's loopback; the proxy is for the
|
||||
// outside world.
|
||||
"NO_PROXY=127.0.0.1,localhost,::1",
|
||||
"no_proxy=127.0.0.1,localhost,::1",
|
||||
}
|
||||
for _, k := range []string{"PATH", "LANG", "LC_ALL", "TZ", "SYSTEMROOT", "WINDIR"} {
|
||||
if v, ok := os.LookupEnv(k); ok {
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
package hostapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/application/repository"
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
"github.com/Tencent/WeKnora/pluginsdk"
|
||||
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
|
||||
)
|
||||
|
||||
func TestTokens(t *testing.T) {
|
||||
iss := NewIssuer([]byte("k"))
|
||||
tok, exp, err := iss.Issue("acme.x", "1.0.0", 7, []string{"kv"})
|
||||
require.NoError(t, err)
|
||||
require.WithinDuration(t, time.Now().Add(TokenTTL), exp, time.Second)
|
||||
c, err := iss.Verify(tok)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, uint64(7), c.TenantID)
|
||||
require.True(t, c.Has("kv"))
|
||||
|
||||
_, err = NewIssuer([]byte("other")).Verify(tok)
|
||||
require.ErrorIs(t, err, ErrInvalidToken, "another key must not verify")
|
||||
later := NewIssuer([]byte("k"))
|
||||
later.now = func() time.Time { return time.Now().Add(TokenTTL + time.Minute) }
|
||||
_, err = later.Verify(tok)
|
||||
require.ErrorIs(t, err, ErrInvalidToken, "expired tokens must not verify")
|
||||
_, err = iss.Verify(tok[:len(tok)-2] + "xx")
|
||||
require.ErrorIs(t, err, ErrInvalidToken)
|
||||
}
|
||||
|
||||
// hostCall is what a plugin's call would carry, pointing at a test server.
|
||||
func hostCall(t *testing.T, url string, iss *Issuer, tenant uint64, scopes ...string) *pluginsdk.Call {
|
||||
t.Helper()
|
||||
tok, _, err := iss.Issue("acme.x", "1.0.0", tenant, scopes)
|
||||
require.NoError(t, err)
|
||||
return &pluginsdk.Call{
|
||||
Context: pluginapi.Context{TenantID: tenant, Host: &pluginapi.HostAccess{URL: url, Token: tok}},
|
||||
}
|
||||
}
|
||||
|
||||
func TestKVThroughTheSDK(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, db.AutoMigrate(&types.PluginKV{}))
|
||||
iss := NewIssuer([]byte("k"))
|
||||
h := NewHandler(iss, NewKV(repository.NewPluginKVRepository(db)))
|
||||
r := gin.New()
|
||||
h.Register(r)
|
||||
srv := httptest.NewServer(r)
|
||||
defer srv.Close()
|
||||
ctx := context.Background()
|
||||
|
||||
host := hostCall(t, srv.URL, iss, 1, "kv").Host()
|
||||
require.NotNil(t, host)
|
||||
type cursor struct{ Page int }
|
||||
ok, err := host.KVGet(ctx, "cursor/a", &cursor{})
|
||||
require.NoError(t, err)
|
||||
require.False(t, ok)
|
||||
require.NoError(t, host.KVPut(ctx, "cursor/a", cursor{Page: 3}, 0))
|
||||
require.NoError(t, host.KVPut(ctx, "cursor/b", cursor{Page: 4}, time.Hour))
|
||||
var got cursor
|
||||
ok, err = host.KVGet(ctx, "cursor/a", &got)
|
||||
require.NoError(t, err)
|
||||
require.True(t, ok)
|
||||
require.Equal(t, 3, got.Page)
|
||||
|
||||
page, err := host.KVList(ctx, "cursor/", "", 1)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, page.Entries, 1)
|
||||
require.Equal(t, "cursor/a", page.Next)
|
||||
page, err = host.KVList(ctx, "cursor/", page.Next, 1)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "cursor/b", page.Entries[0].Key)
|
||||
require.NotNil(t, page.Entries[0].ExpiresAt)
|
||||
|
||||
// Another tenant sees nothing of tenant 1.
|
||||
other := hostCall(t, srv.URL, iss, 2, "kv").Host()
|
||||
ok, err = other.KVGet(ctx, "cursor/a", &got)
|
||||
require.NoError(t, err)
|
||||
require.False(t, ok)
|
||||
|
||||
require.NoError(t, host.KVDelete(ctx, "cursor/a"))
|
||||
ok, _ = host.KVGet(ctx, "cursor/a", &got)
|
||||
require.False(t, ok)
|
||||
|
||||
// Without the scope, or with a forged token, the call is refused.
|
||||
noScope := hostCall(t, srv.URL, iss, 1).Host()
|
||||
_, err = noScope.KVGet(ctx, "cursor/b", &got)
|
||||
require.ErrorContains(t, err, `not granted the "kv" scope`)
|
||||
forged := hostCall(t, srv.URL, NewIssuer([]byte("wrong")), 1, "kv").Host()
|
||||
_, err = forged.KVGet(ctx, "cursor/b", &got)
|
||||
require.ErrorContains(t, err, "invalid or expired")
|
||||
|
||||
// Bad input is a bad request, not a crash.
|
||||
err = host.KVPut(ctx, strings.Repeat("k", 300), 1, 0)
|
||||
e, ok := pluginapi.AsError(err)
|
||||
require.True(t, ok)
|
||||
require.Equal(t, pluginapi.CodeBadRequest, e.Code)
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
package hostapi
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/logger"
|
||||
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
|
||||
)
|
||||
|
||||
// PathPrefix is where the Host API is served. The global auth middleware
|
||||
// lets it through; Handler authenticates plugin tokens itself.
|
||||
const PathPrefix = "/api/v1/plugin-host/"
|
||||
|
||||
const claimsKey = "pluginHostClaims"
|
||||
|
||||
// Handler serves the Host API.
|
||||
type Handler struct {
|
||||
issuer *Issuer
|
||||
kv *KV
|
||||
}
|
||||
|
||||
// NewHandler creates the Host API handler.
|
||||
func NewHandler(issuer *Issuer, kv *KV) *Handler { return &Handler{issuer: issuer, kv: kv} }
|
||||
|
||||
func writeErr(c *gin.Context, err error) {
|
||||
var e *pluginapi.Error
|
||||
if !errors.As(err, &e) {
|
||||
logger.Errorf(c.Request.Context(), "[plugin] host api %s: %v", c.Request.URL.Path, err)
|
||||
e = &pluginapi.Error{Code: pluginapi.CodeInternal, Message: "host api failed"}
|
||||
}
|
||||
c.AbortWithStatusJSON(e.Code.HTTPStatus(), pluginapi.ErrorBody{Error: *e})
|
||||
}
|
||||
|
||||
// authenticate verifies the plugin token and the scope the route needs.
|
||||
func (h *Handler) authenticate(scope string) gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
token, ok := strings.CutPrefix(c.GetHeader("Authorization"), "Bearer ")
|
||||
if !ok || token == "" {
|
||||
writeErr(c, pluginapi.Errorf(pluginapi.CodeUnauthorized, "missing plugin token"))
|
||||
return
|
||||
}
|
||||
claims, err := h.issuer.Verify(token)
|
||||
if err != nil {
|
||||
writeErr(c, pluginapi.Errorf(pluginapi.CodeUnauthorized, "invalid or expired plugin token"))
|
||||
return
|
||||
}
|
||||
if !claims.Has(scope) {
|
||||
writeErr(
|
||||
c,
|
||||
pluginapi.Errorf(pluginapi.CodeUnauthorized, "the plugin was not granted the %q scope", scope),
|
||||
)
|
||||
return
|
||||
}
|
||||
c.Set(claimsKey, claims)
|
||||
c.Next()
|
||||
}
|
||||
}
|
||||
|
||||
func claimsOf(c *gin.Context) *Claims {
|
||||
v, _ := c.Get(claimsKey)
|
||||
claims, _ := v.(*Claims)
|
||||
return claims
|
||||
}
|
||||
|
||||
// Register mounts the Host API on the engine root.
|
||||
func (h *Handler) Register(r gin.IRouter) {
|
||||
g := r.Group(strings.TrimSuffix(PathPrefix, "/"))
|
||||
kv := g.Group("/kv", h.authenticate(ScopeKV))
|
||||
kv.GET("", h.kvGet)
|
||||
kv.PUT("", h.kvPut)
|
||||
kv.DELETE("", h.kvDelete)
|
||||
kv.GET("/list", h.kvList)
|
||||
}
|
||||
|
||||
func (h *Handler) kvGet(c *gin.Context) {
|
||||
e, err := h.kv.Get(c.Request.Context(), claimsOf(c), c.Query("key"))
|
||||
if err != nil {
|
||||
writeErr(c, err)
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, e)
|
||||
}
|
||||
|
||||
func (h *Handler) kvPut(c *gin.Context) {
|
||||
var in pluginapi.KVPut
|
||||
body, err := io.ReadAll(io.LimitReader(c.Request.Body, pluginapi.KVMaxValueBytes+pluginapi.KVMaxKeyBytes+1024))
|
||||
if err == nil {
|
||||
err = json.Unmarshal(body, &in)
|
||||
}
|
||||
if err != nil {
|
||||
writeErr(c, pluginapi.Errorf(pluginapi.CodeBadRequest, "body must be {key, value, ttlSeconds}"))
|
||||
return
|
||||
}
|
||||
e, err := h.kv.Put(c.Request.Context(), claimsOf(c), in)
|
||||
if err != nil {
|
||||
writeErr(c, err)
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, e)
|
||||
}
|
||||
|
||||
func (h *Handler) kvDelete(c *gin.Context) {
|
||||
if err := h.kv.Delete(c.Request.Context(), claimsOf(c), c.Query("key")); err != nil {
|
||||
writeErr(c, err)
|
||||
return
|
||||
}
|
||||
c.Status(http.StatusNoContent)
|
||||
}
|
||||
|
||||
func (h *Handler) kvList(c *gin.Context) {
|
||||
limit, _ := strconv.Atoi(c.Query("limit"))
|
||||
out, err := h.kv.List(c.Request.Context(), claimsOf(c), c.Query("prefix"), c.Query("after"), limit)
|
||||
if err != nil {
|
||||
writeErr(c, err)
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, out)
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
package hostapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"time"
|
||||
"unicode"
|
||||
"unicode/utf8"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/logger"
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
"github.com/Tencent/WeKnora/internal/types/interfaces"
|
||||
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
|
||||
)
|
||||
|
||||
// ScopeKV grants a plugin its own key-value store in the calling tenant.
|
||||
const ScopeKV = "kv"
|
||||
|
||||
// KV is the plugins' key-value store. Every operation is confined to the
|
||||
// plugin and tenant of the token, so a plugin never sees another plugin's
|
||||
// or another workspace's data.
|
||||
type KV struct {
|
||||
repo interfaces.PluginKVRepository
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
// NewKV creates the store.
|
||||
func NewKV(repo interfaces.PluginKVRepository) *KV { return &KV{repo: repo, now: time.Now} }
|
||||
|
||||
func checkKey(key string) error {
|
||||
switch {
|
||||
case key == "":
|
||||
return pluginapi.Errorf(pluginapi.CodeBadRequest, "key is required")
|
||||
case len(key) > pluginapi.KVMaxKeyBytes:
|
||||
return pluginapi.Errorf(pluginapi.CodeBadRequest, "key is over %d bytes", pluginapi.KVMaxKeyBytes)
|
||||
case !utf8.ValidString(key) || strings.ContainsFunc(key, unicode.IsControl):
|
||||
return pluginapi.Errorf(pluginapi.CodeBadRequest, "key must be printable UTF-8")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func toEntry(e *types.PluginKV) pluginapi.KVEntry {
|
||||
return pluginapi.KVEntry{
|
||||
Key: e.Key,
|
||||
Value: json.RawMessage(e.Value),
|
||||
ExpiresAt: e.ExpiresAt,
|
||||
UpdatedAt: e.UpdatedAt,
|
||||
}
|
||||
}
|
||||
|
||||
// Get reads a key.
|
||||
func (s *KV) Get(ctx context.Context, c *Claims, key string) (*pluginapi.KVEntry, error) {
|
||||
if err := checkKey(key); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
e, err := s.repo.Get(ctx, c.PluginID, c.TenantID, key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if e == nil {
|
||||
return nil, pluginapi.Errorf(pluginapi.CodeNotFound, "no key %q", key)
|
||||
}
|
||||
out := toEntry(e)
|
||||
return &out, nil
|
||||
}
|
||||
|
||||
// Put writes a key, refusing new keys past the per-tenant quota.
|
||||
func (s *KV) Put(ctx context.Context, c *Claims, in pluginapi.KVPut) (*pluginapi.KVEntry, error) {
|
||||
if err := checkKey(in.Key); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(in.Value) == 0 || !json.Valid(in.Value) {
|
||||
return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "value must be JSON")
|
||||
}
|
||||
if len(in.Value) > pluginapi.KVMaxValueBytes {
|
||||
return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "value is over %d bytes", pluginapi.KVMaxValueBytes)
|
||||
}
|
||||
if in.TTLSeconds < 0 {
|
||||
return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "ttlSeconds must not be negative")
|
||||
}
|
||||
existing, err := s.repo.Get(ctx, c.PluginID, c.TenantID, in.Key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if existing == nil {
|
||||
n, err := s.repo.Count(ctx, c.PluginID, c.TenantID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if n >= pluginapi.KVMaxKeys {
|
||||
return nil, pluginapi.Errorf(
|
||||
pluginapi.CodeRateLimited,
|
||||
"the store holds its maximum of %d keys",
|
||||
pluginapi.KVMaxKeys,
|
||||
)
|
||||
}
|
||||
}
|
||||
now := s.now()
|
||||
e := &types.PluginKV{
|
||||
PluginID: c.PluginID, TenantID: c.TenantID, Key: in.Key, Value: types.JSON(in.Value), UpdatedAt: now,
|
||||
}
|
||||
if in.TTLSeconds > 0 {
|
||||
exp := now.Add(time.Duration(in.TTLSeconds) * time.Second)
|
||||
e.ExpiresAt = &exp
|
||||
}
|
||||
if err := s.repo.Put(ctx, e); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := toEntry(e)
|
||||
return &out, nil
|
||||
}
|
||||
|
||||
// Delete removes a key; deleting a missing key is not an error.
|
||||
func (s *KV) Delete(ctx context.Context, c *Claims, key string) error {
|
||||
if err := checkKey(key); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err := s.repo.Delete(ctx, c.PluginID, c.TenantID, key)
|
||||
return err
|
||||
}
|
||||
|
||||
// List pages through keys with a prefix.
|
||||
func (s *KV) List(ctx context.Context, c *Claims, prefix, after string, limit int) (*pluginapi.KVList, error) {
|
||||
if limit <= 0 {
|
||||
limit = 100
|
||||
}
|
||||
limit = min(limit, pluginapi.KVMaxListLimit)
|
||||
rows, err := s.repo.List(ctx, c.PluginID, c.TenantID, prefix, after, limit+1)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := &pluginapi.KVList{Entries: []pluginapi.KVEntry{}}
|
||||
for i := range rows {
|
||||
if i == limit {
|
||||
out.Next = rows[i-1].Key
|
||||
break
|
||||
}
|
||||
out.Entries = append(out.Entries, toEntry(&rows[i]))
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// StartSweeper deletes expired entries every interval until ctx ends.
|
||||
func (s *KV) StartSweeper(ctx context.Context, interval time.Duration) {
|
||||
go func() {
|
||||
t := time.NewTicker(interval)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
if n, err := s.repo.DeleteExpired(ctx, s.now()); err != nil {
|
||||
logger.Warnf(ctx, "[plugin] sweep expired kv: %v", err)
|
||||
} else if n > 0 {
|
||||
logger.Infof(ctx, "[plugin] swept %d expired kv entries", n)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
// Package hostapi is the Host API: how a code plugin calls back into
|
||||
// WeKnora. Every call to a plugin that was granted Host API scopes carries a
|
||||
// short-lived token bound to the plugin, the tenant and those scopes; the
|
||||
// plugin presents it to /api/v1/plugin-host/*.
|
||||
package hostapi
|
||||
|
||||
import (
|
||||
"crypto/hmac"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/golang-jwt/jwt/v5"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/utils"
|
||||
)
|
||||
|
||||
// Audience of Host API tokens; they are refused anywhere else.
|
||||
const Audience = "weknora-host-api"
|
||||
|
||||
// TokenTTL is how long a token lives: one call and its follow-ups.
|
||||
const TokenTTL = 5 * time.Minute
|
||||
|
||||
// Claims are what a Host API token asserts.
|
||||
type Claims struct {
|
||||
PluginID string `json:"plugin"`
|
||||
Version string `json:"ver"`
|
||||
TenantID uint64 `json:"tenant"`
|
||||
Scopes []string `json:"scopes"`
|
||||
jwt.RegisteredClaims
|
||||
}
|
||||
|
||||
// Has reports whether the token grants a scope.
|
||||
func (c *Claims) Has(scope string) bool { return slices.Contains(c.Scopes, scope) }
|
||||
|
||||
// Issuer signs and verifies Host API tokens.
|
||||
type Issuer struct {
|
||||
key []byte
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
// NewIssuer signs with key. Nodes must share it for a token issued on one
|
||||
// node to verify on another.
|
||||
func NewIssuer(key []byte) *Issuer { return &Issuer{key: key, now: time.Now} }
|
||||
|
||||
// NewIssuerFromEnv derives the signing key from the cluster-wide secrets:
|
||||
// SYSTEM_AES_KEY, else JWT_SECRET. Without either it uses a random key, which
|
||||
// only works while plugins call back the node that called them (the
|
||||
// embedded host always does).
|
||||
func NewIssuerFromEnv() *Issuer {
|
||||
var secret []byte
|
||||
switch {
|
||||
case utils.GetAESKey() != nil:
|
||||
secret = utils.GetAESKey()
|
||||
case strings.TrimSpace(os.Getenv("JWT_SECRET")) != "":
|
||||
secret = []byte(strings.TrimSpace(os.Getenv("JWT_SECRET")))
|
||||
default:
|
||||
secret = make([]byte, 32)
|
||||
_, _ = rand.Read(secret)
|
||||
}
|
||||
mac := hmac.New(sha256.New, secret)
|
||||
_, _ = mac.Write([]byte("weknora plugin host api tokens"))
|
||||
return NewIssuer(mac.Sum(nil))
|
||||
}
|
||||
|
||||
// Issue signs a token for one call.
|
||||
func (i *Issuer) Issue(pluginID, version string, tenantID uint64, scopes []string) (string, time.Time, error) {
|
||||
now := i.now()
|
||||
exp := now.Add(TokenTTL)
|
||||
claims := Claims{
|
||||
PluginID: pluginID, Version: version, TenantID: tenantID, Scopes: scopes,
|
||||
RegisteredClaims: jwt.RegisteredClaims{
|
||||
Subject: "plugin:" + pluginID,
|
||||
Audience: jwt.ClaimStrings{Audience},
|
||||
IssuedAt: jwt.NewNumericDate(now),
|
||||
NotBefore: jwt.NewNumericDate(now.Add(-30 * time.Second)),
|
||||
ExpiresAt: jwt.NewNumericDate(exp),
|
||||
},
|
||||
}
|
||||
tok, err := jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString(i.key)
|
||||
return tok, exp, err
|
||||
}
|
||||
|
||||
// ErrInvalidToken is any token that does not verify.
|
||||
var ErrInvalidToken = errors.New("invalid plugin token")
|
||||
|
||||
// Verify checks a token's signature, audience and lifetime.
|
||||
func (i *Issuer) Verify(token string) (*Claims, error) {
|
||||
var c Claims
|
||||
parsed, err := jwt.ParseWithClaims(token, &c, func(*jwt.Token) (any, error) {
|
||||
return i.key, nil
|
||||
},
|
||||
jwt.WithValidMethods([]string{jwt.SigningMethodHS256.Alg()}),
|
||||
jwt.WithAudience(Audience),
|
||||
jwt.WithExpirationRequired(),
|
||||
jwt.WithTimeFunc(i.now),
|
||||
)
|
||||
if err != nil || !parsed.Valid {
|
||||
return nil, fmt.Errorf("%w: %v", ErrInvalidToken, err)
|
||||
}
|
||||
if c.PluginID == "" || c.TenantID == 0 || c.Subject != "plugin:"+c.PluginID {
|
||||
return nil, fmt.Errorf("%w: incomplete claims", ErrInvalidToken)
|
||||
}
|
||||
return &c, nil
|
||||
}
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
"golang.org/x/mod/semver"
|
||||
@@ -150,6 +151,8 @@ type Contribution struct {
|
||||
Path string `json:"path,omitempty" yaml:"path"`
|
||||
// MCP describes a remote MCP server (mcpServers).
|
||||
MCP *MCPServer `json:"mcp,omitempty" yaml:"mcp"`
|
||||
// FileTypes are the lower-case extensions a parser handles ("pdf").
|
||||
FileTypes []string `json:"fileTypes,omitempty" yaml:"fileTypes"`
|
||||
// Extra carries point-specific metadata the generic fields do not cover.
|
||||
Extra map[string]any `json:"extra,omitempty" yaml:"extra"`
|
||||
}
|
||||
@@ -287,7 +290,19 @@ var egressPattern = regexp.MustCompile(`^(\*\.)?([a-z0-9]([a-z0-9-]*[a-z0-9])?)(
|
||||
// follows links anywhere). Private addresses stay unreachable either way.
|
||||
const EgressAnyHost = "*"
|
||||
|
||||
// HostAPIScopes are the Host API scopes a plugin may ask for in
|
||||
// permissions.hostApi.
|
||||
var HostAPIScopes = []string{
|
||||
// kv: the plugin's own key-value store in each tenant.
|
||||
"kv",
|
||||
}
|
||||
|
||||
func (m *Manifest) validatePermissions(add func(string, ...any)) {
|
||||
for _, s := range m.Permissions.HostAPI {
|
||||
if !slices.Contains(HostAPIScopes, s) {
|
||||
add("permissions.hostApi %q is not a Host API scope (known: %s)", s, strings.Join(HostAPIScopes, ", "))
|
||||
}
|
||||
}
|
||||
for _, e := range m.Permissions.Egress {
|
||||
if e != EgressAnyHost && !egressPattern.MatchString(strings.ToLower(e)) {
|
||||
add(
|
||||
@@ -345,6 +360,9 @@ func (m *Manifest) validateContributions(add func(string, ...any)) {
|
||||
add("%s.aliases may only be declared by builtin plugins", where)
|
||||
}
|
||||
validateDeclarative(point, c, m.Builtin, where, add)
|
||||
if point == PointParsers && !m.Builtin {
|
||||
validateFileTypes(c.FileTypes, where, add)
|
||||
}
|
||||
for _, alias := range c.Aliases {
|
||||
// Aliases share the ID alphabet, which has no '/', so an
|
||||
// alias can never shadow a qualified ID.
|
||||
@@ -358,6 +376,20 @@ func (m *Manifest) validateContributions(add func(string, ...any)) {
|
||||
|
||||
// validateDeclarative checks the fields a declarative contribution needs.
|
||||
// Builtins describe their implementation in code instead.
|
||||
// fileTypePattern is a lower-case file extension without the dot.
|
||||
var fileTypePattern = regexp.MustCompile(`^[a-z0-9]{1,16}$`)
|
||||
|
||||
func validateFileTypes(types []string, where string, add func(string, ...any)) {
|
||||
if len(types) == 0 {
|
||||
add("%s.fileTypes is required: the extensions the parser handles, such as [pdf]", where)
|
||||
}
|
||||
for _, t := range types {
|
||||
if !fileTypePattern.MatchString(t) {
|
||||
add("%s.fileTypes %q must be a lower-case extension without the dot", where, t)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func validateDeclarative(point Point, c Contribution, builtin bool, where string, add func(string, ...any)) {
|
||||
if builtin {
|
||||
return
|
||||
|
||||
@@ -40,7 +40,7 @@ var points = []PointInfo{
|
||||
{Point: PointIMChannels},
|
||||
{Point: PointWebSearch, ThirdParty: true},
|
||||
{Point: PointTools},
|
||||
{Point: PointParsers},
|
||||
{Point: PointParsers, ThirdParty: true},
|
||||
{Point: PointSkills, ThirdParty: true, Declarative: true},
|
||||
{Point: PointMCPServers, ThirdParty: true, Declarative: true},
|
||||
}
|
||||
|
||||
@@ -20,6 +20,7 @@ import (
|
||||
"github.com/Tencent/WeKnora/internal/logger"
|
||||
"github.com/Tencent/WeKnora/internal/mcpserver"
|
||||
"github.com/Tencent/WeKnora/internal/middleware"
|
||||
"github.com/Tencent/WeKnora/internal/plugin/hostapi"
|
||||
"github.com/Tencent/WeKnora/internal/tracing/langfuse"
|
||||
"github.com/Tencent/WeKnora/internal/types/interfaces"
|
||||
|
||||
@@ -92,6 +93,7 @@ type RouterParams struct {
|
||||
DataSourceCredentialsHandler *handler.DataSourceCredentialsHandler
|
||||
PluginHandler *handler.PluginHandler
|
||||
PluginAdminHandler *handler.PluginAdminHandler
|
||||
PluginHostAPI *hostapi.Handler
|
||||
WeKnoraCloudHandler *handler.WeKnoraCloudHandler
|
||||
WikiPageHandler *handler.WikiPageHandler
|
||||
MemoryHandler *handler.MemoryHandler
|
||||
@@ -206,6 +208,12 @@ func NewRouter(params RouterParams) *gin.Engine {
|
||||
r.POST("/api/v1/local-browser/extension/authorize", params.SessionHandler.BrowserSkillAuthorize)
|
||||
r.POST("/api/v1/local-browser/internal", params.SessionHandler.BrowserSkillInternal)
|
||||
|
||||
// Host API: plugins call back with their own short-lived tokens, which
|
||||
// the handler verifies; user and API key auth do not apply.
|
||||
if params.PluginHostAPI != nil {
|
||||
params.PluginHostAPI.Register(r)
|
||||
}
|
||||
|
||||
// 认证中间件
|
||||
r.Use(middleware.Auth(params.TenantService, params.UserService, params.TenantMemberService, params.TenantAPIKeyService, params.Config))
|
||||
|
||||
|
||||
@@ -51,6 +51,10 @@ type ParserEngineInfo struct {
|
||||
FileTypes []string
|
||||
Available bool
|
||||
UnavailableReason string
|
||||
// PluginID and DisplayNames describe an engine an installed plugin
|
||||
// provides; DisplayNames is keyed by locale.
|
||||
PluginID string `json:",omitempty"`
|
||||
DisplayNames map[string]string `json:",omitempty"`
|
||||
}
|
||||
|
||||
// --- Internal types used by chunking pipeline ---
|
||||
|
||||
@@ -2,6 +2,7 @@ package interfaces
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/plugin/manifest"
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
@@ -41,3 +42,26 @@ type PluginRepository interface {
|
||||
GetVersion(ctx context.Context, pluginID, version string) (*types.PluginVersion, error)
|
||||
SaveVersion(ctx context.Context, v *types.PluginVersion) error
|
||||
}
|
||||
|
||||
// PluginKVRepository stores plugins' Host API key-value data. Expired
|
||||
// entries read as missing.
|
||||
type PluginKVRepository interface {
|
||||
// Get returns (nil, nil) for a missing or expired key.
|
||||
Get(ctx context.Context, pluginID string, tenantID uint64, key string) (*types.PluginKV, error)
|
||||
Put(ctx context.Context, e *types.PluginKV) error
|
||||
// Delete reports whether the key existed.
|
||||
Delete(ctx context.Context, pluginID string, tenantID uint64, key string) (bool, error)
|
||||
// List returns live keys starting with prefix, after the given key, in
|
||||
// key order.
|
||||
List(
|
||||
ctx context.Context,
|
||||
pluginID string,
|
||||
tenantID uint64,
|
||||
prefix, after string,
|
||||
limit int,
|
||||
) ([]types.PluginKV, error)
|
||||
// Count counts live keys of a plugin in a tenant.
|
||||
Count(ctx context.Context, pluginID string, tenantID uint64) (int64, error)
|
||||
// DeleteExpired removes entries that expired before now.
|
||||
DeleteExpired(ctx context.Context, now time.Time) (int64, error)
|
||||
}
|
||||
|
||||
@@ -48,3 +48,17 @@ type PluginVersion struct {
|
||||
|
||||
// TableName pins the table so GORM's pluralizer cannot drift.
|
||||
func (PluginVersion) TableName() string { return "plugin_versions" }
|
||||
|
||||
// PluginKV is one entry of a plugin's key-value store (Host API kv),
|
||||
// partitioned by plugin and tenant.
|
||||
type PluginKV struct {
|
||||
PluginID string `json:"-" gorm:"type:varchar(128);primaryKey"`
|
||||
TenantID uint64 `json:"-" gorm:"primaryKey"`
|
||||
Key string `json:"key" gorm:"type:varchar(256);primaryKey"`
|
||||
Value JSON `json:"value" gorm:"type:json"`
|
||||
ExpiresAt *time.Time `json:"expiresAt,omitempty"`
|
||||
UpdatedAt time.Time `json:"updatedAt"`
|
||||
}
|
||||
|
||||
// TableName pins the table so GORM's pluralizer cannot drift.
|
||||
func (PluginKV) TableName() string { return "plugin_kv" }
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS plugin_kv;
|
||||
@@ -0,0 +1,12 @@
|
||||
-- Host API key-value store of plugins (versioned 000117).
|
||||
CREATE TABLE IF NOT EXISTS plugin_kv (
|
||||
plugin_id VARCHAR(128) NOT NULL,
|
||||
tenant_id BIGINT NOT NULL,
|
||||
key VARCHAR(256) NOT NULL,
|
||||
value TEXT NOT NULL,
|
||||
expires_at DATETIME,
|
||||
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
PRIMARY KEY (plugin_id, tenant_id, key)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_plugin_kv_expires_at ON plugin_kv (expires_at);
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS plugin_kv;
|
||||
@@ -0,0 +1,18 @@
|
||||
-- Migration 000117: the key-value store plugins use through the Host API.
|
||||
-- Plugins keep no database of their own; their state lives here, partitioned
|
||||
-- by plugin and tenant so a plugin cannot read another workspace's data.
|
||||
DO $$ BEGIN RAISE NOTICE '[Migration 000117] Creating plugin_kv'; END $$;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS plugin_kv (
|
||||
plugin_id VARCHAR(128) NOT NULL,
|
||||
tenant_id BIGINT NOT NULL,
|
||||
key VARCHAR(256) NOT NULL,
|
||||
value JSONB NOT NULL,
|
||||
expires_at TIMESTAMPTZ,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
PRIMARY KEY (plugin_id, tenant_id, key)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_plugin_kv_expires_at ON plugin_kv (expires_at) WHERE expires_at IS NOT NULL;
|
||||
|
||||
COMMENT ON TABLE plugin_kv IS 'Host API key-value store of plugins, per plugin and tenant.';
|
||||
+29
-3
@@ -1,7 +1,7 @@
|
||||
# WeKnora plugin SDK (Go)
|
||||
|
||||
Build code plugins for WeKnora: web search providers and data source
|
||||
connectors that run as their own process. The module has no dependencies.
|
||||
Build code plugins for WeKnora: web search providers, data source
|
||||
connectors and document parsers that run as their own process. The module has no dependencies.
|
||||
|
||||
| Package | What it is |
|
||||
| --- | --- |
|
||||
@@ -69,6 +69,31 @@ decrypted, in three separate scopes:
|
||||
|
||||
Keep plugins stateless and never cache a tenant's credentials across calls.
|
||||
|
||||
## Parsers
|
||||
|
||||
`p.Parser(id, ...)` turns a document (`in.Content`, base64 on the wire) into
|
||||
Markdown. WeKnora chunks the text and stores the images the Markdown
|
||||
references: return them in `Images` with an `OriginalRef` matching the
|
||||
`` target. Declare the extensions it handles in `plugin.yaml`
|
||||
(`fileTypes: [srt, vtt]`). Return a retryable error (`unavailable`,
|
||||
`rate_limited`) to have the document retried later; any other error fails
|
||||
it for good.
|
||||
|
||||
## Calling back into WeKnora (Host API)
|
||||
|
||||
A plugin that declares `permissions.hostApi` gets a short-lived token with
|
||||
each call; `call.Host()` returns a client for it, or nil without a grant.
|
||||
Scopes:
|
||||
|
||||
- `kv`: a key-value store of the plugin, separate per workspace, so the
|
||||
plugin needs no database.
|
||||
- `KVGet`, `KVPut` (optional TTL), `KVDelete` and `KVList`.
|
||||
- Keys up to 256 bytes, JSON values up to 64 KB, 10,000 keys per
|
||||
workspace.
|
||||
|
||||
The token is valid for a few minutes: use `call.Host()` within the call and
|
||||
don't keep it.
|
||||
|
||||
## Packaging
|
||||
|
||||
A package is a zip (`.wkp`) with `plugin.yaml` at its root:
|
||||
@@ -103,7 +128,8 @@ A few rules the manifest and host enforce:
|
||||
(`acme.notes/notes`); for connectors and web search that must stay within
|
||||
50 characters.
|
||||
|
||||
See `examples/plugins/rss` for a complete plugin and its `package.sh`.
|
||||
See `examples/plugins/rss` (a connector) and `examples/plugins/subtitles` (a
|
||||
parser using the Host API) for complete plugins with their `package.sh`.
|
||||
|
||||
## Testing
|
||||
|
||||
|
||||
@@ -140,6 +140,15 @@ func Run(ctx context.Context, t Target) Report {
|
||||
return protocolAnswer(err)
|
||||
})
|
||||
}
|
||||
for _, id := range m.Contributes["parsers"] {
|
||||
check("parsers/"+id+" parse answers", func(ctx context.Context) error {
|
||||
var out pluginapi.ParseOutput
|
||||
err := t.Client.Call(ctx, pluginapi.ParsePath(id), envelope(), pluginapi.ParseInput{
|
||||
FileName: "conformance.txt", FileType: "txt", Content: []byte("WeKnora conformance check\n"),
|
||||
}, &out)
|
||||
return protocolAnswer(err)
|
||||
})
|
||||
}
|
||||
for _, id := range m.Contributes["connectors"] {
|
||||
check("connectors/"+id+" validate answers", func(ctx context.Context) error {
|
||||
return protocolAnswer(t.Client.Call(ctx, pluginapi.ConnectorValidatePath(id), envelope(), nil, nil))
|
||||
@@ -227,7 +236,13 @@ func wantErrorBody(resp *http.Response, code pluginapi.ErrorCode) error {
|
||||
return fmt.Errorf("HTTP %d without a protocol error body: %s", resp.StatusCode, b)
|
||||
}
|
||||
if resp.StatusCode != code.HTTPStatus() || body.Error.Code != code {
|
||||
return fmt.Errorf("want HTTP %d %s, got HTTP %d %s", code.HTTPStatus(), code, resp.StatusCode, body.Error.Code)
|
||||
return fmt.Errorf(
|
||||
"want HTTP %d %s, got HTTP %d %s",
|
||||
code.HTTPStatus(),
|
||||
code,
|
||||
resp.StatusCode,
|
||||
body.Error.Code,
|
||||
)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -65,12 +65,25 @@ func TestSDKPluginConforms(t *testing.T) {
|
||||
},
|
||||
),
|
||||
)
|
||||
p.Parser(
|
||||
"text",
|
||||
pluginsdk.ParserFunc(
|
||||
func(_ context.Context, _ *pluginsdk.Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
|
||||
return &pluginapi.ParseOutput{Markdown: string(in.Content)}, nil
|
||||
},
|
||||
),
|
||||
)
|
||||
rep := conformance.Run(context.Background(), target(t, p.Handler()))
|
||||
names := map[string]bool{}
|
||||
for _, r := range rep.Results {
|
||||
names[r.Name] = true
|
||||
if !r.Passed {
|
||||
t.Errorf("%s: %s", r.Name, r.Detail)
|
||||
}
|
||||
}
|
||||
if !names["parsers/text parse answers"] {
|
||||
t.Error("parsers must be checked")
|
||||
}
|
||||
if !rep.Passed() || rep.Plugin != "acme.strict@1.0.0" || len(rep.Results) < 10 {
|
||||
t.Fatalf("report = %+v", rep)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,114 @@
|
||||
package pluginsdk
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
|
||||
)
|
||||
|
||||
// Host calls back into WeKnora on behalf of one call: the tenant and scopes
|
||||
// are those the call's token carries. It is only valid while the token lives
|
||||
// (a few minutes), so do not keep it beyond the call.
|
||||
type Host struct {
|
||||
base string
|
||||
token string
|
||||
http *http.Client
|
||||
}
|
||||
|
||||
// Host returns the Host API client of this call, or nil when the plugin was
|
||||
// granted no Host API scopes.
|
||||
func (c *Call) Host() *Host {
|
||||
if c.Context.Host == nil || c.Context.Host.URL == "" || c.Context.Host.Token == "" {
|
||||
return nil
|
||||
}
|
||||
return &Host{
|
||||
base: strings.TrimSuffix(c.Context.Host.URL, "/"), token: c.Context.Host.Token,
|
||||
http: &http.Client{Timeout: 30 * time.Second},
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Host) do(ctx context.Context, method, path string, query url.Values, body, out any) error {
|
||||
var rd io.Reader
|
||||
if body != nil {
|
||||
b, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
rd = bytes.NewReader(b)
|
||||
}
|
||||
u := h.base + path
|
||||
if len(query) > 0 {
|
||||
u += "?" + query.Encode()
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, method, u, rd)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+h.token)
|
||||
if body != nil {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
resp, err := h.http.Do(req)
|
||||
if err != nil {
|
||||
return pluginapi.Errorf(pluginapi.CodeUnavailable, "host api: %v", err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
if resp.StatusCode/100 != 2 {
|
||||
var eb pluginapi.ErrorBody
|
||||
b, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
|
||||
if json.Unmarshal(b, &eb) == nil && eb.Error.Code != "" {
|
||||
return &eb.Error
|
||||
}
|
||||
return pluginapi.Errorf(pluginapi.CodeInternal, "host api answered HTTP %d", resp.StatusCode)
|
||||
}
|
||||
if out == nil || resp.StatusCode == http.StatusNoContent {
|
||||
return nil
|
||||
}
|
||||
return json.NewDecoder(resp.Body).Decode(out)
|
||||
}
|
||||
|
||||
// KVGet reads key into v. ok is false when the key does not exist.
|
||||
func (h *Host) KVGet(ctx context.Context, key string, v any) (ok bool, err error) {
|
||||
var e pluginapi.KVEntry
|
||||
err = h.do(ctx, http.MethodGet, pluginapi.HostKVPath, url.Values{"key": {key}}, nil, &e)
|
||||
if pe, isPE := pluginapi.AsError(err); isPE && pe.Code == pluginapi.CodeNotFound {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, json.Unmarshal(e.Value, v)
|
||||
}
|
||||
|
||||
// KVPut stores v (as JSON) under key; ttl 0 keeps it until deleted.
|
||||
func (h *Host) KVPut(ctx context.Context, key string, v any, ttl time.Duration) error {
|
||||
raw, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
in := pluginapi.KVPut{Key: key, Value: raw, TTLSeconds: int(ttl / time.Second)}
|
||||
return h.do(ctx, http.MethodPut, pluginapi.HostKVPath, nil, in, nil)
|
||||
}
|
||||
|
||||
// KVDelete removes key; a missing key is not an error.
|
||||
func (h *Host) KVDelete(ctx context.Context, key string) error {
|
||||
return h.do(ctx, http.MethodDelete, pluginapi.HostKVPath, url.Values{"key": {key}}, nil, nil)
|
||||
}
|
||||
|
||||
// KVList returns a page of keys with prefix after the given key.
|
||||
func (h *Host) KVList(ctx context.Context, prefix, after string, limit int) (*pluginapi.KVList, error) {
|
||||
q := url.Values{"prefix": {prefix}, "after": {after}}
|
||||
if limit > 0 {
|
||||
q.Set("limit", strconv.Itoa(limit))
|
||||
}
|
||||
var out pluginapi.KVList
|
||||
return &out, h.do(ctx, http.MethodGet, pluginapi.HostKVListPath, q, nil, &out)
|
||||
}
|
||||
@@ -25,7 +25,7 @@ func TestOpenAPIMatchesRoutes(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, f := range []string{"websearch.go", "connector.go"} {
|
||||
for _, f := range []string{"websearch.go", "connector.go", "parser.go"} {
|
||||
b, err := os.ReadFile(f)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
package pluginsdk
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
|
||||
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
|
||||
)
|
||||
|
||||
// Parser turns a document into Markdown. Return a non-retryable error (such
|
||||
// as pluginapi.CodeInvalidConfig or CodeNotFound) for a document that will
|
||||
// never parse; retryable ones (CodeUnavailable, CodeRateLimited) make
|
||||
// WeKnora try again later.
|
||||
type Parser interface {
|
||||
Parse(ctx context.Context, call *Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error)
|
||||
}
|
||||
|
||||
// ParserFunc adapts a function to Parser.
|
||||
type ParserFunc func(ctx context.Context, call *Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error)
|
||||
|
||||
// Parse implements Parser.
|
||||
func (f ParserFunc) Parse(ctx context.Context, call *Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
|
||||
return f(ctx, call, in)
|
||||
}
|
||||
|
||||
// Parser registers parser id (contributes.parsers[].id).
|
||||
func (p *Plugin) Parser(id string, parser Parser) { p.parsers[id] = parser }
|
||||
|
||||
func (p *Plugin) routeParsers(mux *http.ServeMux) {
|
||||
mux.HandleFunc("POST /v1/parsers/{id}/parse", func(w http.ResponseWriter, r *http.Request) {
|
||||
parser, ok := p.parsers[r.PathValue("id")]
|
||||
if !ok {
|
||||
writeError(w, pluginapi.Errorf(pluginapi.CodeNotFound, "no parser %q", r.PathValue("id")))
|
||||
return
|
||||
}
|
||||
p.unary(func(ctx context.Context, call *Call, raw json.RawMessage) (any, error) {
|
||||
in, err := decodeInput[pluginapi.ParseInput](raw)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(in.Content) == 0 && in.URL == "" {
|
||||
return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "content or url is required")
|
||||
}
|
||||
out, err := parser.Parse(ctx, call, in)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if out == nil {
|
||||
out = &pluginapi.ParseOutput{}
|
||||
}
|
||||
return out, nil
|
||||
})(w, r)
|
||||
})
|
||||
}
|
||||
@@ -62,6 +62,7 @@ type Plugin struct {
|
||||
info Info
|
||||
webSearch map[string]WebSearcher
|
||||
connectors map[string]Connector
|
||||
parsers map[string]Parser
|
||||
validate ConfigValidator
|
||||
logger *slog.Logger
|
||||
// ShutdownTimeout bounds how long Serve waits for calls in flight after
|
||||
@@ -75,6 +76,7 @@ func New(info Info) *Plugin {
|
||||
info: info,
|
||||
webSearch: map[string]WebSearcher{},
|
||||
connectors: map[string]Connector{},
|
||||
parsers: map[string]Parser{},
|
||||
logger: slog.New(slog.NewTextHandler(os.Stderr, nil)),
|
||||
ShutdownTimeout: 60 * time.Second,
|
||||
}
|
||||
@@ -100,6 +102,7 @@ func (p *Plugin) Manifest() pluginapi.Manifest {
|
||||
}
|
||||
add("webSearch", keys(p.webSearch))
|
||||
add("connectors", keys(p.connectors))
|
||||
add("parsers", keys(p.parsers))
|
||||
return m
|
||||
}
|
||||
|
||||
@@ -134,6 +137,7 @@ func (p *Plugin) Handler() http.Handler {
|
||||
)
|
||||
p.routeWebSearch(mux)
|
||||
p.routeConnectors(mux)
|
||||
p.routeParsers(mux)
|
||||
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, pluginapi.Errorf(pluginapi.CodeNotFound, "no endpoint %s %s", r.Method, r.URL.Path))
|
||||
})
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
package pluginapi
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Host API: how a plugin calls back into WeKnora. The URL and a short-lived
|
||||
// bearer token arrive in Context.Host of each call; the token is bound to the
|
||||
// tenant of that call and the scopes the plugin was granted (manifest
|
||||
// permissions.hostApi). Errors use ErrorBody.
|
||||
|
||||
// HostKVPath is the key-value endpoint (scope "kv"): GET and DELETE take
|
||||
// ?key=, PUT takes a KVPut body.
|
||||
const HostKVPath = "/api/v1/plugin-host/kv"
|
||||
|
||||
// HostKVListPath lists keys: ?prefix=&after=&limit=.
|
||||
const HostKVListPath = "/api/v1/plugin-host/kv/list"
|
||||
|
||||
// KVEntry is one stored value.
|
||||
type KVEntry struct {
|
||||
Key string `json:"key"`
|
||||
Value json.RawMessage `json:"value"`
|
||||
ExpiresAt *time.Time `json:"expiresAt,omitempty"`
|
||||
UpdatedAt time.Time `json:"updatedAt"`
|
||||
}
|
||||
|
||||
// KVPut stores a value; TTLSeconds 0 keeps it until deleted.
|
||||
type KVPut struct {
|
||||
Key string `json:"key"`
|
||||
Value json.RawMessage `json:"value"`
|
||||
TTLSeconds int `json:"ttlSeconds,omitempty"`
|
||||
}
|
||||
|
||||
// KVList is a page of keys in key order; pass Next as after for the next
|
||||
// page (empty when there is none).
|
||||
type KVList struct {
|
||||
Entries []KVEntry `json:"entries"`
|
||||
Next string `json:"next,omitempty"`
|
||||
}
|
||||
|
||||
// Limits of the key-value store.
|
||||
const (
|
||||
KVMaxKeyBytes = 256
|
||||
KVMaxValueBytes = 64 << 10
|
||||
KVMaxKeys = 10000 // per plugin and tenant
|
||||
KVMaxListLimit = 1000
|
||||
)
|
||||
@@ -162,6 +162,33 @@ paths:
|
||||
application/x-ndjson:
|
||||
schema: { $ref: "#/components/schemas/Event" }
|
||||
default: { $ref: "#/components/responses/Error" }
|
||||
/v1/parsers/{id}/parse:
|
||||
post:
|
||||
summary: Parse a document into Markdown (DocReader.Read)
|
||||
description: |
|
||||
The file arrives as base64 content (or a URL). Answer Markdown and
|
||||
the images it references; WeKnora stores the images and chunks the
|
||||
text. A non-retryable error fails the document for good; a
|
||||
retryable one (unavailable, rate_limited) is retried later.
|
||||
parameters: [{ $ref: "#/components/parameters/ContributionID" }]
|
||||
requestBody:
|
||||
required: true
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
allOf:
|
||||
- $ref: "#/components/schemas/Envelope"
|
||||
- properties: { input: { $ref: "#/components/schemas/ParseInput" } }
|
||||
responses:
|
||||
"200":
|
||||
description: The parsed document
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
required: [output]
|
||||
properties: { output: { $ref: "#/components/schemas/ParseOutput" } }
|
||||
default: { $ref: "#/components/responses/Error" }
|
||||
components:
|
||||
securitySchemes:
|
||||
hostToken: { type: http, scheme: bearer }
|
||||
@@ -296,6 +323,29 @@ components:
|
||||
metadata: { type: object, additionalProperties: { type: string } }
|
||||
isDeleted: { type: boolean }
|
||||
sourceResourceId: { type: string }
|
||||
ParseInput:
|
||||
type: object
|
||||
properties:
|
||||
fileName: { type: string }
|
||||
fileType: { type: string, description: "Lower-case extension without the dot" }
|
||||
content: { type: string, contentEncoding: base64 }
|
||||
url: { type: string }
|
||||
title: { type: string }
|
||||
ParseOutput:
|
||||
type: object
|
||||
required: [markdown]
|
||||
properties:
|
||||
markdown: { type: string }
|
||||
images:
|
||||
type: array
|
||||
items:
|
||||
type: object
|
||||
required: [originalRef, data]
|
||||
properties:
|
||||
originalRef: { type: string }
|
||||
mimeType: { type: string }
|
||||
data: { type: string, contentEncoding: base64 }
|
||||
metadata: { type: object, additionalProperties: { type: string } }
|
||||
Event:
|
||||
type: object
|
||||
required: [type]
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
package pluginapi
|
||||
|
||||
// ParsePath is the parse endpoint of parser id.
|
||||
func ParsePath(id string) string { return "/v1/parsers/" + id + "/parse" }
|
||||
|
||||
// ParseInput is one document to parse: its bytes (file mode) or a URL (URL
|
||||
// mode). The parser's own settings arrive in Config.System and Config.Tenant.
|
||||
type ParseInput struct {
|
||||
FileName string `json:"fileName,omitempty"`
|
||||
// FileType is the lower-case extension without the dot ("pdf").
|
||||
FileType string `json:"fileType,omitempty"`
|
||||
// Content is the file; JSON carries it as base64.
|
||||
Content []byte `json:"content,omitempty"`
|
||||
URL string `json:"url,omitempty"`
|
||||
Title string `json:"title,omitempty"`
|
||||
}
|
||||
|
||||
// ParseOutput is the document as Markdown. WeKnora chunks it; parsers do not.
|
||||
type ParseOutput struct {
|
||||
Markdown string `json:"markdown"`
|
||||
// Images are the pictures the Markdown references: an image whose
|
||||
// OriginalRef equals a  in Markdown is stored and the link
|
||||
// rewritten. Inline data: URIs in the Markdown work too.
|
||||
Images []ParsedImage `json:"images,omitempty"`
|
||||
// Metadata such as "title" (URL mode) or "pages".
|
||||
Metadata map[string]string `json:"metadata,omitempty"`
|
||||
}
|
||||
|
||||
// ParsedImage is one image of a parsed document.
|
||||
type ParsedImage struct {
|
||||
OriginalRef string `json:"originalRef"`
|
||||
MimeType string `json:"mimeType,omitempty"`
|
||||
Data []byte `json:"data"`
|
||||
}
|
||||
Reference in New Issue
Block a user