From 9ca2bdf239372d5ecc11e32f96c20f5d30f57316 Mon Sep 17 00:00:00 2001 From: Darko Atanasovski Date: Sun, 28 Jun 2026 11:42:33 +0200 Subject: [PATCH] msgpack support --- .gitignore | 4 +- app/components/key/content.tsx | 3 +- app/components/key/header.tsx | 4 +- app/lib/utils.ts | 42 ++++++ pkg/handlers/get.go | 3 +- pkg/handlers/scan.go | 4 +- pkg/utils/msgpack.go | 258 ++++++++++++++++++++++++++++++++- pkg/utils/msgpack_test.go | 173 +++++++++++++++++++++- 8 files changed, 473 insertions(+), 18 deletions(-) diff --git a/.gitignore b/.gitignore index 0f0c38c..a190302 100644 --- a/.gitignore +++ b/.gitignore @@ -6,11 +6,11 @@ !.yarn/versions node_modules - +.DS_Store #!.yarn/cache .pnp.* .bin/ -bin/ \ No newline at end of file +bin/ diff --git a/app/components/key/content.tsx b/app/components/key/content.tsx index 91a3d37..4dcb58d 100644 --- a/app/components/key/content.tsx +++ b/app/components/key/content.tsx @@ -1,5 +1,6 @@ import { ScrollArea } from "@/components/ui/scroll-area"; import { ScanItem } from "@/hooks/use-keys"; +import { formatRawValue } from "@/lib/utils"; import { Prism as SyntaxHighlighter } from "react-syntax-highlighter"; import { vscDarkPlus } from "react-syntax-highlighter/dist/esm/styles/prism"; @@ -29,7 +30,7 @@ export function KeyDetailsContent({ {JSON.stringify(selectedItem.value, null, 2)} ) : ( - selectedItem.raw_value + formatRawValue(selectedItem.raw_value) )} diff --git a/app/components/key/header.tsx b/app/components/key/header.tsx index 29af3f0..8c5789f 100644 --- a/app/components/key/header.tsx +++ b/app/components/key/header.tsx @@ -7,7 +7,7 @@ import { KeyIcon, TrashIcon, } from "lucide-react"; -import { formatBytes } from "@/lib/utils"; +import { formatBytes, rawValueByteLength } from "@/lib/utils"; interface KeyDetailsHeaderProps { selectedItem: ScanItem | null; @@ -30,7 +30,7 @@ export function KeyDetailsHeader({ {selectedItem.key} - ({formatBytes(selectedItem.raw_value.length)}) + ({formatBytes(rawValueByteLength(selectedItem.raw_value))}) diff --git a/app/lib/utils.ts b/app/lib/utils.ts index 98d6e5d..fdca954 100644 --- a/app/lib/utils.ts +++ b/app/lib/utils.ts @@ -5,6 +5,48 @@ export function cn(...inputs: ClassValue[]) { return twMerge(clsx(inputs)); } +/** Decode API raw_value (plain text or base64-encoded binary) to a display string. */ +export function formatRawValue(raw: string): string { + if (!raw) return ""; + + try { + const binary = atob(raw); + const bytes = Uint8Array.from(binary, (c) => c.charCodeAt(0)); + const hex = Array.from(bytes) + .map((b) => b.toString(16).padStart(2, "0")) + .join(" "); + + let text = ""; + for (const b of bytes) { + if (b >= 0x20 && b <= 0x7e) { + text += String.fromCharCode(b); + } else if (b === 0x0a) { + text += "\\n"; + } else if (b === 0x0d) { + text += "\\r"; + } else if (b === 0x09) { + text += "\\t"; + } else { + text += "."; + } + } + + return `hex: ${hex}\n\nascii: ${text}`; + } catch { + return raw; + } +} + +/** Byte length of an API raw_value string. */ +export function rawValueByteLength(raw: string): number { + if (!raw) return 0; + try { + return atob(raw).length; + } catch { + return new TextEncoder().encode(raw).length; + } +} + export function formatBytes(bytes: number, decimals = 2) { if (!+bytes) return "0 Bytes"; diff --git a/pkg/handlers/get.go b/pkg/handlers/get.go index 0441d91..20884d4 100644 --- a/pkg/handlers/get.go +++ b/pkg/handlers/get.go @@ -42,7 +42,8 @@ func Get(s *server.Server) http.HandlerFunc { Key: req.Key, } if val != nil { - resp.Value, resp.RawValue = utils.ParseValue(val) + resp.Value, _ = utils.ParseValue(val) + resp.RawValue = utils.FormatRawValue(val) resp.Found = true } diff --git a/pkg/handlers/scan.go b/pkg/handlers/scan.go index adc05eb..30710e3 100644 --- a/pkg/handlers/scan.go +++ b/pkg/handlers/scan.go @@ -40,11 +40,11 @@ func Scan(s *server.Server) http.HandlerFunc { items := make([]types.ScanItem, 0, len(keys)) for i := range keys { - parsed, raw := utils.ParseValue(values[i]) + parsed, _ := utils.ParseValue(values[i]) items = append(items, types.ScanItem{ Key: string(keys[i]), Value: parsed, - RawValue: raw, + RawValue: utils.FormatRawValue(values[i]), }) } diff --git a/pkg/utils/msgpack.go b/pkg/utils/msgpack.go index d96f190..7a7ddd4 100644 --- a/pkg/utils/msgpack.go +++ b/pkg/utils/msgpack.go @@ -1,15 +1,57 @@ package utils import ( + "bytes" + "encoding/base64" + "encoding/binary" + "encoding/hex" "encoding/json" + "fmt" + "io" + "strconv" + "sync" "unicode/utf8" "github.com/vmihailenco/msgpack/v5" + "github.com/vmihailenco/msgpack/v5/msgpcode" ) +// msgpackExt handles Stream feeds extension type 0 (e.g. the compact "c" timestamp field). +type msgpackExt struct { + Payload []byte +} + +func (e *msgpackExt) MarshalMsgpack() ([]byte, error) { + return e.Payload, nil +} + +func (e *msgpackExt) UnmarshalMsgpack(b []byte) error { + e.Payload = append([]byte(nil), b...) + return nil +} + +var registerMsgpackExt sync.Once + +func ensureMsgpackExt() { + registerMsgpackExt.Do(func() { + // Feeds TiKV values use msgpack ext types for compact binary fields (e.g. "c", "v"). + for extID := int8(0); extID < 32; extID++ { + msgpack.RegisterExt(extID, (*msgpackExt)(nil)) + } + }) +} + +// FormatRawValue returns a JSON-safe representation of raw TiKV bytes. +func FormatRawValue(data []byte) string { + if isPlainText(data) { + return string(data) + } + return base64.StdEncoding.EncodeToString(data) +} + // ParseValue attempts to parse a byte slice as msgpack / JSON and return a structured value. func ParseValue(data []byte) (parsed any, raw string) { - raw = string(data) + raw = FormatRawValue(data) if isPlainText(data) { var decoded any @@ -20,8 +62,8 @@ func ParseValue(data []byte) (parsed any, raw string) { return raw, raw } - var decoded any - if err := msgpack.Unmarshal(data, &decoded); err != nil { + decoded, err := decodeMsgpack(data) + if err != nil { if err := json.Unmarshal(data, &decoded); err != nil { return raw, raw } @@ -37,7 +79,215 @@ func ParseValue(data []byte) (parsed any, raw string) { } } - return decoded, raw + return unwrapVersioned(decoded), raw +} + +func decodeMsgpack(data []byte) (any, error) { + ensureMsgpackExt() + + values, err := decodeMsgpackValues(data) + if err != nil { + return nil, err + } + + switch len(values) { + case 0: + return nil, io.EOF + case 1: + return values[0], nil + default: + return values, nil + } +} + +func decodeMsgpackValues(data []byte) ([]any, error) { + dec := msgpack.NewDecoder(bytes.NewReader(data)) + dec.UsePreallocateValues(true) + + var values []any + for { + var v any + err := dec.Decode(&v) + if err == io.EOF { + break + } + if err != nil { + return nil, err + } + values = append(values, normalizeDecoded(v)) + } + return values, nil +} + +// unwrapVersioned strips the feeds storage version prefix: a leading 0 followed by the payload. +func unwrapVersioned(v any) any { + arr, ok := v.([]any) + if !ok || len(arr) != 2 { + return v + } + if version, ok := arr[0].(int64); ok && version == 0 { + return arr[1] + } + if version, ok := arr[0].(int8); ok && version == 0 { + return arr[1] + } + if version, ok := arr[0].(uint8); ok && version == 0 { + return arr[1] + } + if version, ok := arr[0].(int); ok && version == 0 { + return arr[1] + } + return v +} + +func normalizeDecoded(v any) any { + switch x := v.(type) { + case map[string]any: + out := make(map[string]any, len(x)) + for k, val := range x { + out[k] = normalizeDecoded(val) + } + return out + case map[any]any: + out := make(map[string]any, len(x)) + for k, val := range x { + out[normalizeMapKey(k)] = normalizeDecoded(val) + } + return out + case []any: + out := make([]any, len(x)) + for i, val := range x { + out[i] = normalizeDecoded(val) + } + return out + case *msgpackExt: + if x == nil { + return nil + } + return extToJSON(0, x.Payload) + case msgpackExt: + return extToJSON(0, x.Payload) + case []byte: + if nested, ok := tryDecodeNestedMsgpack(x); ok { + return nested + } + return bytesToJSON(x) + default: + return v + } +} + +// tryDecodeNestedMsgpack decodes bin fields that embed msgpack values (e.g. feeds "p"). +func tryDecodeNestedMsgpack(data []byte) (any, bool) { + if !looksLikeEmbeddedMsgpack(data) { + return nil, false + } + + values, err := decodeMsgpackValues(data) + if err != nil || len(values) == 0 { + return nil, false + } + if len(values) == 1 { + return values[0], true + } + return values, true +} + +func looksLikeEmbeddedMsgpack(data []byte) bool { + if len(data) < 2 { + return false + } + + c := data[0] + if msgpcode.IsFixedMap(c) { + return true + } + if msgpcode.IsFixedString(c) { + return true + } + if msgpcode.IsFixedArray(c) && c > msgpcode.FixedArrayLow { + return true + } + + switch c { + case msgpcode.Map16, msgpcode.Map32, + msgpcode.Array16, msgpcode.Array32, + msgpcode.Str8, msgpcode.Str16, msgpcode.Str32, + msgpcode.Bin8, msgpcode.Bin16, msgpcode.Bin32, + msgpcode.FixExt1, msgpcode.FixExt2, msgpcode.FixExt4, msgpcode.FixExt8, msgpcode.FixExt16, + msgpcode.Ext8, msgpcode.Ext16, msgpcode.Ext32: + return true + } + + return false +} + +func bytesToJSON(b []byte) map[string]any { + return map[string]any{ + "$hex": hex.EncodeToString(b), + "$len": len(b), + } +} + +func normalizeMapKey(k any) string { + switch key := k.(type) { + case string: + return key + case []byte: + return string(key) + case int: + return strconv.Itoa(key) + case int8: + return strconv.FormatInt(int64(key), 10) + case int16: + return strconv.FormatInt(int64(key), 10) + case int32: + return strconv.FormatInt(int64(key), 10) + case int64: + return strconv.FormatInt(key, 10) + case uint: + return strconv.FormatUint(uint64(key), 10) + case uint8: + return strconv.FormatUint(uint64(key), 10) + case uint16: + return strconv.FormatUint(uint64(key), 10) + case uint32: + return strconv.FormatUint(uint64(key), 10) + case uint64: + return strconv.FormatUint(key, 10) + default: + return fmt.Sprint(key) + } +} + +func extToJSON(extID int8, payload []byte) any { + if ts, ok := decodeExtTimestamp(payload); ok { + return ts + } + return map[string]any{ + "$ext": extID, + "$hex": hex.EncodeToString(payload), + "$len": len(payload), + } +} + +// decodeExtTimestamp interprets 8-byte feeds ext payloads as nanosecond timestamps when plausible. +func decodeExtTimestamp(payload []byte) (int64, bool) { + if len(payload) != 8 { + return 0, false + } + for _, n := range []uint64{ + binary.BigEndian.Uint64(payload), + binary.LittleEndian.Uint64(payload), + } { + if n >= 1e17 && n <= 9e18 { + ts := int64(n) + if ts > 0 { + return ts, true + } + } + } + return 0, false } // isPlainText checks if the data appears to be plain text (UTF-8, mostly printable) diff --git a/pkg/utils/msgpack_test.go b/pkg/utils/msgpack_test.go index 32d3278..42ffbd8 100644 --- a/pkg/utils/msgpack_test.go +++ b/pkg/utils/msgpack_test.go @@ -1,7 +1,10 @@ package utils import ( + "encoding/base64" "testing" + + "github.com/vmihailenco/msgpack/v5" ) func TestParseValue(t *testing.T) { @@ -33,7 +36,7 @@ func TestParseValue(t *testing.T) { name: "JSON String", input: []byte(`"quoted string"`), wantRaw: `"quoted string"`, - wantJSON: true, // It is a valid JSON string, so it should be parsed as string + wantJSON: true, }, { name: "Spacing", @@ -55,16 +58,12 @@ func TestParseValue(t *testing.T) { } if tt.wantJSON { - // Assert that parsed is NOT the raw string (unless the JSON value IS a string) - // But specifically for object/array it should be map/slice switch parsed.(type) { case map[string]any, []any, string, float64, bool, nil: - // OK default: t.Errorf("ParseValue() parsed type = %T, want JSON type", parsed) } - // For "Simple JSON Object" specifically: if tt.name == "Simple JSON Object" { m, ok := parsed.(map[string]any) if !ok || m["foo"] != "bar" { @@ -72,7 +71,6 @@ func TestParseValue(t *testing.T) { } } } else { - // For plain text, parsed should equal raw if parsed != raw { t.Errorf("ParseValue() parsed = %v, want %v", parsed, raw) } @@ -80,3 +78,166 @@ func TestParseValue(t *testing.T) { }) } } + +func TestFormatRawValue_Base64(t *testing.T) { + data := []byte{0x00, 0x01, 0xff} + want := base64.StdEncoding.EncodeToString(data) + if got := FormatRawValue(data); got != want { + t.Fatalf("FormatRawValue() = %q, want %q", got, want) + } +} + +func makeFeedActivity() map[string]any { + return map[string]any{ + "fid": "user:test", + "op": int8(1), + "a": "$2b454049-e5bd-4efb-9c55-395beb3ad026", + "u": "test", + "o": "user:test", + "v": int8(1), + "c": &msgpackExt{Payload: []byte{0x18, 0, 0, 0, 0x59, 0, 0, 0}}, + "ag": []string{"fg:notification:0", "_post_2026-05-22", "fg:stories:0", "test"}, + "at": "post", + } +} + +func TestParseValue_FeedsMsgpackActivity(t *testing.T) { + ensureMsgpackExt() + + data, err := msgpack.Marshal(makeFeedActivity()) + if err != nil { + t.Fatal(err) + } + + parsed, _ := ParseValue(data) + m, ok := parsed.(map[string]any) + if !ok { + t.Fatalf("expected map, got %T", parsed) + } + if m["fid"] != "user:test" { + t.Fatalf("fid = %v", m["fid"]) + } + if m["at"] != "post" { + t.Fatalf("at = %v", m["at"]) + } + + switch m["c"].(type) { + case int64, int, uint64, map[string]any: + default: + t.Fatalf("unexpected c type %T: %v", m["c"], m["c"]) + } +} + +func TestParseValue_FeedsMsgpackActivityArray(t *testing.T) { + ensureMsgpackExt() + + activities := []any{makeFeedActivity(), makeFeedActivity()} + data, err := msgpack.Marshal(activities) + if err != nil { + t.Fatal(err) + } + + parsed, _ := ParseValue(data) + arr, ok := parsed.([]any) + if !ok { + t.Fatalf("expected array, got %T", parsed) + } + if len(arr) != 2 { + t.Fatalf("len = %d", len(arr)) + } +} + +func TestParseValue_FeedsNestedPayload(t *testing.T) { + ensureMsgpackExt() + + ts := []byte{0x18, 0, 0, 0, 0x59, 0x4d, 0x60, 0x00} + inner, err := msgpack.Marshal(map[string]any{ + "a": "test", + "c": &msgpackExt{Payload: ts}, + "g": "user", + }) + if err != nil { + t.Fatal(err) + } + + data, err := msgpack.Marshal(map[string]any{ + "c": &msgpackExt{Payload: ts}, + "p": inner, + "a": "test", + "g": "user", + "v": &msgpackExt{Payload: []byte{0xd0, 0x90, 0xd4, 0x87, 0x0c}}, + "i": []byte{0x01, 0x02, 0x03}, + }) + if err != nil { + t.Fatal(err) + } + + parsed, _ := ParseValue(data) + m, ok := parsed.(map[string]any) + if !ok { + t.Fatalf("expected map, got %T", parsed) + } + + p, ok := m["p"].(map[string]any) + if !ok { + t.Fatalf("expected nested map in p, got %T: %v", m["p"], m["p"]) + } + if p["a"] != "test" || p["g"] != "user" { + t.Fatalf("nested p = %v", p) + } + + i, ok := m["i"].(map[string]any) + if !ok { + t.Fatalf("expected hex wrapper for opaque i, got %T", m["i"]) + } + if i["$hex"] != "010203" { + t.Fatalf("i hex = %v", i["$hex"]) + } +} + +func TestParseValue_ConcatenatedMsgpackMaps(t *testing.T) { + ensureMsgpackExt() + + a1, err := msgpack.Marshal(makeFeedActivity()) + if err != nil { + t.Fatal(err) + } + a2, err := msgpack.Marshal(makeFeedActivity()) + if err != nil { + t.Fatal(err) + } + data := append(a1, a2...) + + parsed, _ := ParseValue(data) + arr, ok := parsed.([]any) + if !ok { + t.Fatalf("expected array of maps, got %T", parsed) + } + if len(arr) != 2 { + t.Fatalf("len = %d", len(arr)) + } +} + +func TestParseValue_VersionedWrapper(t *testing.T) { + ensureMsgpackExt() + + inner, err := msgpack.Marshal(map[string]any{ + "fid": "user:test", + "op": int8(1), + "at": "post", + }) + if err != nil { + t.Fatal(err) + } + // Feeds storage: leading version 0 (fixint) + msgpack payload. + data := append([]byte{0x00}, inner...) + + parsed, _ := ParseValue(data) + m, ok := parsed.(map[string]any) + if !ok { + t.Fatalf("expected unwrapped map, got %T: %v", parsed, parsed) + } + if m["fid"] != "user:test" { + t.Fatalf("fid = %v", m["fid"]) + } +}