atscan.net
[DEPRECATED] Go implementation of plcbundle
perf updates
tree.fail
benchmark_detector
benchmark_detector
This is a binary file and will not be displayed.
+2
-2
bundle/bundle_test.go
+2
-2
bundle/bundle_test.go
+2
-1
bundle/clone.go
+2
-1
bundle/clone.go
+2
-1
bundle/index.go
+2
-1
bundle/index.go
+2
-1
bundle/mempool.go
+2
-1
bundle/mempool.go
+26
-65
bundle/operations.go
+26
-65
bundle/operations.go
···
5
"bytes"
6
"crypto/sha256"
7
"encoding/hex"
8
-
"encoding/json"
9
"fmt"
10
"io"
11
"os"
12
"time"
13
14
-
"github.com/klauspost/compress/zstd"
15
"tangled.org/atscan.net/plcbundle/plc"
16
)
17
18
// Operations handles low-level bundle file operations
19
type Operations struct {
20
-
encoder *zstd.Encoder
21
-
decoder *zstd.Decoder
22
-
logger Logger
23
}
24
25
-
// NewOperations creates a new Operations handler with default compression
26
func NewOperations(logger Logger) (*Operations, error) {
27
-
// Always use default compression (level 3 - good balance)
28
-
encoder, err := zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedDefault))
29
-
if err != nil {
30
-
return nil, fmt.Errorf("failed to create zstd encoder: %w", err)
31
-
}
32
-
33
-
decoder, err := zstd.NewReader(nil)
34
-
if err != nil {
35
-
return nil, fmt.Errorf("failed to create zstd decoder: %w", err)
36
-
}
37
-
38
-
return &Operations{
39
-
encoder: encoder,
40
-
decoder: decoder,
41
-
logger: logger,
42
-
}, nil
43
}
44
45
-
// Close cleans up resources
46
func (op *Operations) Close() {
47
-
if op.encoder != nil {
48
-
op.encoder.Close()
49
-
}
50
-
if op.decoder != nil {
51
-
op.decoder.Close()
52
-
}
53
}
54
55
// ========================================
···
81
func (op *Operations) ParseJSONL(data []byte) ([]plc.PLCOperation, error) {
82
var operations []plc.PLCOperation
83
scanner := bufio.NewScanner(bytes.NewReader(data))
84
-
85
-
// Set a large buffer for long lines
86
buf := make([]byte, 0, 64*1024)
87
scanner.Buffer(buf, 1024*1024)
88
89
-
lineNum := 0
90
for scanner.Scan() {
91
-
lineNum++
92
line := scanner.Bytes()
93
-
94
if len(line) == 0 {
95
continue
96
}
97
98
var operation plc.PLCOperation
99
if err := json.Unmarshal(line, &operation); err != nil {
100
-
return nil, fmt.Errorf("failed to parse line %d: %w", lineNum, err)
101
}
102
103
-
// Store raw JSON
104
operation.RawJSON = make([]byte, len(line))
105
copy(operation.RawJSON, line)
106
-
107
operations = append(operations, operation)
108
}
109
110
-
if err := scanner.Err(); err != nil {
111
-
return nil, fmt.Errorf("scanner error: %w", err)
112
-
}
113
-
114
return operations, nil
115
}
116
···
118
// FILE OPERATIONS (uses JSONL + compression)
119
// ========================================
120
121
-
// LoadBundle loads a compressed bundle from disk
122
func (op *Operations) LoadBundle(path string) ([]plc.PLCOperation, error) {
123
-
// Read compressed file
124
compressed, err := os.ReadFile(path)
125
if err != nil {
126
return nil, fmt.Errorf("failed to read file: %w", err)
127
}
128
129
-
// Decompress
130
-
decompressed, err := op.decoder.DecodeAll(compressed, nil)
131
if err != nil {
132
return nil, fmt.Errorf("failed to decompress: %w", err)
133
}
134
135
-
// Parse JSONL
136
return op.ParseJSONL(decompressed)
137
}
138
139
// SaveBundle saves operations to disk (compressed)
140
// Returns: contentHash, compressedHash, contentSize, compressedSize, error
141
func (op *Operations) SaveBundle(path string, operations []plc.PLCOperation) (string, string, int64, int64, error) {
142
-
// Serialize to JSONL
143
jsonlData := op.SerializeJSONL(operations)
144
contentSize := int64(len(jsonlData))
145
contentHash := op.Hash(jsonlData)
146
147
-
// Compress
148
-
compressed := op.encoder.EncodeAll(jsonlData, nil)
149
compressedSize := int64(len(compressed))
150
compressedHash := op.Hash(compressed)
151
152
-
// Write to file
153
if err := os.WriteFile(path, compressed, 0644); err != nil {
154
return "", "", 0, 0, fmt.Errorf("failed to write file: %w", err)
155
}
···
177
return nil, fmt.Errorf("failed to open bundle: %w", err)
178
}
179
180
-
// Create a new decoder for this stream
181
-
decoder, err := zstd.NewReader(file)
182
-
if err != nil {
183
-
file.Close()
184
-
return nil, fmt.Errorf("failed to create decompressor: %w", err)
185
-
}
186
187
-
// Return a wrapper that closes both the decoder and file
188
return &decompressedReader{
189
-
decoder: decoder,
190
-
file: file,
191
}, nil
192
}
193
194
// decompressedReader wraps a zstd decoder and underlying file
195
type decompressedReader struct {
196
-
decoder *zstd.Decoder
197
-
file *os.File
198
}
199
200
func (dr *decompressedReader) Read(p []byte) (int, error) {
201
-
return dr.decoder.Read(p)
202
}
203
204
func (dr *decompressedReader) Close() error {
205
-
dr.decoder.Close()
206
return dr.file.Close()
207
}
208
···
241
compressedHash = op.Hash(compressedData)
242
compressedSize = int64(len(compressedData))
243
244
-
// Decompress
245
-
decompressed, err := op.decoder.DecodeAll(compressedData, nil)
246
if err != nil {
247
return "", 0, "", 0, fmt.Errorf("failed to decompress: %w", err)
248
}
···
5
"bytes"
6
"crypto/sha256"
7
"encoding/hex"
8
"fmt"
9
"io"
10
"os"
11
"time"
12
13
+
gozstd "github.com/DataDog/zstd"
14
+
"github.com/goccy/go-json"
15
"tangled.org/atscan.net/plcbundle/plc"
16
)
17
18
// Operations handles low-level bundle file operations
19
type Operations struct {
20
+
logger Logger
21
}
22
23
func NewOperations(logger Logger) (*Operations, error) {
24
+
return &Operations{logger: logger}, nil
25
}
26
27
func (op *Operations) Close() {
28
+
// Nothing to close
29
}
30
31
// ========================================
···
57
func (op *Operations) ParseJSONL(data []byte) ([]plc.PLCOperation, error) {
58
var operations []plc.PLCOperation
59
scanner := bufio.NewScanner(bytes.NewReader(data))
60
buf := make([]byte, 0, 64*1024)
61
scanner.Buffer(buf, 1024*1024)
62
63
for scanner.Scan() {
64
line := scanner.Bytes()
65
if len(line) == 0 {
66
continue
67
}
68
69
var operation plc.PLCOperation
70
+
// Use sonic instead of json.Unmarshal
71
if err := json.Unmarshal(line, &operation); err != nil {
72
+
return nil, fmt.Errorf("failed to parse line: %w", err)
73
}
74
75
operation.RawJSON = make([]byte, len(line))
76
copy(operation.RawJSON, line)
77
operations = append(operations, operation)
78
}
79
80
return operations, nil
81
}
82
···
84
// FILE OPERATIONS (uses JSONL + compression)
85
// ========================================
86
87
+
// LoadBundle loads a compressed bundle
88
func (op *Operations) LoadBundle(path string) ([]plc.PLCOperation, error) {
89
compressed, err := os.ReadFile(path)
90
if err != nil {
91
return nil, fmt.Errorf("failed to read file: %w", err)
92
}
93
94
+
decompressed, err := gozstd.Decompress(nil, compressed)
95
if err != nil {
96
return nil, fmt.Errorf("failed to decompress: %w", err)
97
}
98
99
return op.ParseJSONL(decompressed)
100
}
101
102
// SaveBundle saves operations to disk (compressed)
103
// Returns: contentHash, compressedHash, contentSize, compressedSize, error
104
func (op *Operations) SaveBundle(path string, operations []plc.PLCOperation) (string, string, int64, int64, error) {
105
jsonlData := op.SerializeJSONL(operations)
106
contentSize := int64(len(jsonlData))
107
contentHash := op.Hash(jsonlData)
108
109
+
// DataDog zstd.Compress returns ([]byte, error)
110
+
compressed, err := gozstd.Compress(nil, jsonlData)
111
+
if err != nil {
112
+
return "", "", 0, 0, fmt.Errorf("failed to compress: %w", err)
113
+
}
114
+
115
compressedSize := int64(len(compressed))
116
compressedHash := op.Hash(compressed)
117
118
if err := os.WriteFile(path, compressed, 0644); err != nil {
119
return "", "", 0, 0, fmt.Errorf("failed to write file: %w", err)
120
}
···
142
return nil, fmt.Errorf("failed to open bundle: %w", err)
143
}
144
145
+
// Create zstd reader using DataDog's package
146
+
reader := gozstd.NewReader(file)
147
148
+
// Return a wrapper that closes both the reader and file
149
return &decompressedReader{
150
+
reader: reader,
151
+
file: file,
152
}, nil
153
}
154
155
// decompressedReader wraps a zstd decoder and underlying file
156
type decompressedReader struct {
157
+
reader io.ReadCloser
158
+
file *os.File
159
}
160
161
func (dr *decompressedReader) Read(p []byte) (int, error) {
162
+
return dr.reader.Read(p)
163
}
164
165
func (dr *decompressedReader) Close() error {
166
+
dr.reader.Close()
167
return dr.file.Close()
168
}
169
···
202
compressedHash = op.Hash(compressedData)
203
compressedSize = int64(len(compressedData))
204
205
+
// Decompress with DataDog zstd
206
+
decompressed, err := gozstd.Decompress(nil, compressedData)
207
if err != nil {
208
return "", 0, "", 0, fmt.Errorf("failed to decompress: %w", err)
209
}
+2
-1
cmd/plcbundle/compare.go
+2
-1
cmd/plcbundle/compare.go
+17
-1
cmd/plcbundle/detector.go
+17
-1
cmd/plcbundle/detector.go
···
4
import (
5
"bufio"
6
"context"
7
-
"encoding/json"
8
"flag"
9
"fmt"
10
"os"
11
"sort"
12
"strings"
13
14
"tangled.org/atscan.net/plcbundle/detector"
15
"tangled.org/atscan.net/plcbundle/plc"
···
311
fs := flag.NewFlagSet("detector run", flag.ExitOnError)
312
bundleRange := fs.String("bundles", "", "bundle range, default: all bundles")
313
confidence := fs.Float64("confidence", 0.90, "minimum confidence")
314
fs.Parse(flagArgs)
315
316
// Load manager
317
mgr, _, err := getManager("")
···
4
import (
5
"bufio"
6
"context"
7
"flag"
8
"fmt"
9
+
"net/http"
10
+
_ "net/http/pprof"
11
"os"
12
"sort"
13
"strings"
14
+
"time"
15
+
16
+
"github.com/goccy/go-json"
17
18
"tangled.org/atscan.net/plcbundle/detector"
19
"tangled.org/atscan.net/plcbundle/plc"
···
315
fs := flag.NewFlagSet("detector run", flag.ExitOnError)
316
bundleRange := fs.String("bundles", "", "bundle range, default: all bundles")
317
confidence := fs.Float64("confidence", 0.90, "minimum confidence")
318
+
pprofPort := fs.String("pprof", "", "enable pprof on port (e.g., :6060)")
319
fs.Parse(flagArgs)
320
+
321
+
// Start pprof server if requested
322
+
if *pprofPort != "" {
323
+
go func() {
324
+
fmt.Fprintf(os.Stderr, "pprof server starting on http://localhost%s/debug/pprof/\n", *pprofPort)
325
+
if err := http.ListenAndServe(*pprofPort, nil); err != nil {
326
+
fmt.Fprintf(os.Stderr, "pprof server failed: %v\n", err)
327
+
}
328
+
}()
329
+
time.Sleep(100 * time.Millisecond) // Let server start
330
+
}
331
332
// Load manager
333
mgr, _, err := getManager("")
+5
-1
cmd/plcbundle/main.go
+5
-1
cmd/plcbundle/main.go
···
2
3
import (
4
"context"
5
"flag"
6
"fmt"
7
"net/http"
···
15
"sync"
16
"syscall"
17
"time"
18
+
19
+
"github.com/goccy/go-json"
20
21
"tangled.org/atscan.net/plcbundle/bundle"
22
"tangled.org/atscan.net/plcbundle/plc"
···
56
}
57
58
func main() {
59
+
60
+
debug.SetGCPercent(400)
61
+
62
if len(os.Args) < 2 {
63
printUsage()
64
os.Exit(1)
+2
-1
cmd/plcbundle/server.go
+2
-1
cmd/plcbundle/server.go
+42
-8
detector/builtin.go
+42
-8
detector/builtin.go
···
49
func (d *InvalidHandleDetector) Version() string { return "1.0.0" }
50
51
func (d *InvalidHandleDetector) Detect(ctx context.Context, op plc.PLCOperation) (*Match, error) {
52
-
if alsoKnownAs, ok := op.Operation["alsoKnownAs"].([]interface{}); ok {
53
for _, aka := range alsoKnownAs {
54
if str, ok := aka.(string); ok {
55
// Check if it's an at:// handle
···
203
func (d *AlsoKnownAsSpamDetector) Version() string { return "1.0.0" }
204
205
func (d *AlsoKnownAsSpamDetector) Detect(ctx context.Context, op plc.PLCOperation) (*Match, error) {
206
-
if alsoKnownAs, ok := op.Operation["alsoKnownAs"].([]interface{}); ok {
207
entryCount := len(alsoKnownAs)
208
209
// Count different types of entries
···
296
func (d *SpamPDSDetector) Version() string { return "1.0.0" }
297
298
func (d *SpamPDSDetector) Detect(ctx context.Context, op plc.PLCOperation) (*Match, error) {
299
// Check PDS endpoint
300
-
if services, ok := op.Operation["services"].(map[string]interface{}); ok {
301
if pds, ok := services["atproto_pds"].(map[string]interface{}); ok {
302
if endpoint, ok := pds["endpoint"].(string); ok {
303
host := extractHost(endpoint)
···
320
}
321
322
// Check for spam domain claims in alsoKnownAs
323
-
if alsoKnownAs, ok := op.Operation["alsoKnownAs"].([]interface{}); ok {
324
for _, aka := range alsoKnownAs {
325
if str, ok := aka.(string); ok {
326
if !strings.HasPrefix(str, "at://") {
···
391
func (d *ServiceAbuseDetector) Version() string { return "1.0.0" }
392
393
func (d *ServiceAbuseDetector) Detect(ctx context.Context, op plc.PLCOperation) (*Match, error) {
394
-
if services, ok := op.Operation["services"].(map[string]interface{}); ok {
395
// Check for numeric service keys (spam uses "0", "1", "2" instead of proper names)
396
hasNumericKeys := false
397
numericKeyCount := 0
···
457
}
458
459
// Check for excessively long handles in alsoKnownAs
460
-
if alsoKnownAs, ok := op.Operation["alsoKnownAs"].([]interface{}); ok {
461
for _, aka := range alsoKnownAs {
462
if str, ok := aka.(string); ok {
463
if strings.HasPrefix(str, "at://") {
···
480
}
481
482
// Check for empty verificationMethods (common in this spam)
483
-
if vm, ok := op.Operation["verificationMethods"].(map[string]interface{}); ok {
484
if len(vm) == 0 {
485
// Empty verificationMethods alone isn't enough, but combined with other signals...
486
// Check if there are other suspicious signals
487
-
if services, ok := op.Operation["services"].(map[string]interface{}); ok {
488
if len(services) > 2 {
489
// Multiple services + empty verificationMethods = suspicious
490
return &Match{
···
49
func (d *InvalidHandleDetector) Version() string { return "1.0.0" }
50
51
func (d *InvalidHandleDetector) Detect(ctx context.Context, op plc.PLCOperation) (*Match, error) {
52
+
// Parse Operation field on-demand
53
+
operation, err := op.GetOperationMap()
54
+
if err != nil {
55
+
return nil, err
56
+
}
57
+
if operation == nil {
58
+
return nil, nil
59
+
}
60
+
61
+
if alsoKnownAs, ok := operation["alsoKnownAs"].([]interface{}); ok {
62
+
63
for _, aka := range alsoKnownAs {
64
if str, ok := aka.(string); ok {
65
// Check if it's an at:// handle
···
213
func (d *AlsoKnownAsSpamDetector) Version() string { return "1.0.0" }
214
215
func (d *AlsoKnownAsSpamDetector) Detect(ctx context.Context, op plc.PLCOperation) (*Match, error) {
216
+
// Parse Operation field on-demand
217
+
operation, err := op.GetOperationMap()
218
+
if err != nil {
219
+
return nil, err
220
+
}
221
+
if operation == nil {
222
+
return nil, nil
223
+
}
224
+
if alsoKnownAs, ok := operation["alsoKnownAs"].([]interface{}); ok {
225
entryCount := len(alsoKnownAs)
226
227
// Count different types of entries
···
314
func (d *SpamPDSDetector) Version() string { return "1.0.0" }
315
316
func (d *SpamPDSDetector) Detect(ctx context.Context, op plc.PLCOperation) (*Match, error) {
317
+
// Parse Operation field on-demand
318
+
operation, err := op.GetOperationMap()
319
+
if err != nil {
320
+
return nil, err
321
+
}
322
+
if operation == nil {
323
+
return nil, nil
324
+
}
325
// Check PDS endpoint
326
+
if services, ok := operation["services"].(map[string]interface{}); ok {
327
if pds, ok := services["atproto_pds"].(map[string]interface{}); ok {
328
if endpoint, ok := pds["endpoint"].(string); ok {
329
host := extractHost(endpoint)
···
346
}
347
348
// Check for spam domain claims in alsoKnownAs
349
+
if alsoKnownAs, ok := operation["alsoKnownAs"].([]interface{}); ok {
350
for _, aka := range alsoKnownAs {
351
if str, ok := aka.(string); ok {
352
if !strings.HasPrefix(str, "at://") {
···
417
func (d *ServiceAbuseDetector) Version() string { return "1.0.0" }
418
419
func (d *ServiceAbuseDetector) Detect(ctx context.Context, op plc.PLCOperation) (*Match, error) {
420
+
// Parse Operation field on-demand
421
+
operation, err := op.GetOperationMap()
422
+
if err != nil {
423
+
return nil, err
424
+
}
425
+
if operation == nil {
426
+
return nil, nil
427
+
}
428
+
if services, ok := operation["services"].(map[string]interface{}); ok {
429
// Check for numeric service keys (spam uses "0", "1", "2" instead of proper names)
430
hasNumericKeys := false
431
numericKeyCount := 0
···
491
}
492
493
// Check for excessively long handles in alsoKnownAs
494
+
if alsoKnownAs, ok := operation["alsoKnownAs"].([]interface{}); ok {
495
for _, aka := range alsoKnownAs {
496
if str, ok := aka.(string); ok {
497
if strings.HasPrefix(str, "at://") {
···
514
}
515
516
// Check for empty verificationMethods (common in this spam)
517
+
if vm, ok := operation["verificationMethods"].(map[string]interface{}); ok {
518
if len(vm) == 0 {
519
// Empty verificationMethods alone isn't enough, but combined with other signals...
520
// Check if there are other suspicious signals
521
+
if services, ok := operation["services"].(map[string]interface{}); ok {
522
if len(services) > 2 {
523
// Multiple services + empty verificationMethods = suspicious
524
return &Match{
+2
-1
detector/script.go
+2
-1
detector/script.go
+4
-7
go.mod
+4
-7
go.mod
···
1
module tangled.org/atscan.net/plcbundle
2
3
-
go 1.23
4
-
5
-
require github.com/klauspost/compress v1.18.1
6
-
7
-
require github.com/gorilla/websocket v1.5.3
8
9
require (
10
-
github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
11
-
github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
12
)
+4
-6
go.sum
+4
-6
go.sum
···
1
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
2
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
3
-
github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co=
4
-
github.com/klauspost/compress v1.18.1/go.mod h1:ZQFFVG+MdnR0P+l6wpXgIL4NTtwiKIdBnrBd8Nrxr+0=
5
-
github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=
6
-
github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
7
-
github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
8
-
github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
···
1
+
github.com/DataDog/zstd v1.5.7 h1:ybO8RBeh29qrxIhCA9E8gKY6xfONU9T6G6aP9DTKfLE=
2
+
github.com/DataDog/zstd v1.5.7/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
3
+
github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
4
+
github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
5
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
6
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+2
-1
plc/client.go
+2
-1
plc/client.go
+2
-2
plc/plc_test.go
+2
-2
plc/plc_test.go
···
94
DID: "did:plc:test" + string(rune(i)),
95
CID: "bafytest" + string(rune(i)),
96
CreatedAt: time.Now(),
97
-
Operation: map[string]interface{}{"type": "create"},
98
}
99
json.NewEncoder(w).Encode(op)
100
}
···
239
DID: "did:plc:test",
240
CID: "bafytest",
241
CreatedAt: time.Now(),
242
-
Operation: map[string]interface{}{"type": "create"},
243
}
244
}
245
···
94
DID: "did:plc:test" + string(rune(i)),
95
CID: "bafytest" + string(rune(i)),
96
CreatedAt: time.Now(),
97
+
//Operation: map[string]interface{}{"type": "create"},
98
}
99
json.NewEncoder(w).Encode(op)
100
}
···
239
DID: "did:plc:test",
240
CID: "bafytest",
241
CreatedAt: time.Now(),
242
+
//Operation: map[string]interface{}{"type": "create"},
243
}
244
}
245
+23
-6
plc/types.go
+23
-6
plc/types.go
···
1
package plc
2
3
-
import "time"
4
5
// PLCOperation represents a single operation from the PLC directory
6
type PLCOperation struct {
7
-
DID string `json:"did"`
8
-
Operation map[string]interface{} `json:"operation"`
9
-
CID string `json:"cid"`
10
-
Nullified interface{} `json:"nullified,omitempty"`
11
-
CreatedAt time.Time `json:"createdAt"`
12
13
// RawJSON stores the original JSON bytes for exact reproduction
14
RawJSON []byte `json:"-"`
···
86
Type string // "pds", "labeler", etc.
87
Endpoint string
88
}
···
1
package plc
2
3
+
import (
4
+
"time"
5
+
6
+
"github.com/goccy/go-json"
7
+
)
8
9
// PLCOperation represents a single operation from the PLC directory
10
type PLCOperation struct {
11
+
DID string `json:"did"`
12
+
//Operation map[string]interface{} `json:"operation"`
13
+
Operation json.RawMessage `json:"operation"`
14
+
CID string `json:"cid"`
15
+
Nullified interface{} `json:"nullified,omitempty"`
16
+
CreatedAt time.Time `json:"createdAt"`
17
18
// RawJSON stores the original JSON bytes for exact reproduction
19
RawJSON []byte `json:"-"`
···
91
Type string // "pds", "labeler", etc.
92
Endpoint string
93
}
94
+
95
+
// GetOperationMap parses Operation RawMessage into a map
96
+
func (op *PLCOperation) GetOperationMap() (map[string]interface{}, error) {
97
+
if len(op.Operation) == 0 {
98
+
return nil, nil
99
+
}
100
+
var result map[string]interface{}
101
+
if err := json.Unmarshal(op.Operation, &result); err != nil {
102
+
return nil, err
103
+
}
104
+
return result, nil
105
+
}
+160
scripts/benchmark-detector.go
+160
scripts/benchmark-detector.go
···
···
1
+
package main
2
+
3
+
import (
4
+
"bufio"
5
+
"bytes"
6
+
"fmt"
7
+
"os"
8
+
"path/filepath"
9
+
"strconv"
10
+
"strings"
11
+
"time"
12
+
13
+
gozstd "github.com/DataDog/zstd"
14
+
"github.com/goccy/go-json"
15
+
)
16
+
17
+
// Minimal operation struct
18
+
type Operation struct {
19
+
DID string `json:"did"`
20
+
//Operation map[string]interface{} `json:"operation"`
21
+
CID string `json:"cid"`
22
+
Nullified interface{} `json:"nullified,omitempty"`
23
+
CreatedAt time.Time `json:"createdAt"`
24
+
25
+
// RawJSON stores the original JSON bytes for exact reproduction
26
+
RawJSON []byte `json:"-"`
27
+
}
28
+
29
+
// User's detect function
30
+
func detect(op *Operation) []string {
31
+
labels := []string{}
32
+
33
+
if strings.HasPrefix(op.DID, "did:plc:aa") {
34
+
labels = append(labels, "test")
35
+
}
36
+
37
+
return labels
38
+
}
39
+
40
+
func main() {
41
+
bundleDir := "./"
42
+
if len(os.Args) > 1 {
43
+
bundleDir = os.Args[1]
44
+
}
45
+
46
+
startBundle := 1
47
+
if len(os.Args) > 2 {
48
+
startBundle, _ = strconv.Atoi(os.Args[2])
49
+
}
50
+
51
+
endBundle := 100
52
+
if len(os.Args) > 3 {
53
+
endBundle, _ = strconv.Atoi(os.Args[3])
54
+
}
55
+
56
+
fmt.Fprintf(os.Stderr, "Processing bundles %d-%d from %s\n", startBundle, endBundle, bundleDir)
57
+
fmt.Fprintf(os.Stderr, "\n")
58
+
59
+
// Buffered stdout
60
+
writer := bufio.NewWriterSize(os.Stdout, 512*1024)
61
+
defer writer.Flush()
62
+
63
+
// CSV header
64
+
writer.WriteString("bundle,position,cid,size,confidence,labels\n")
65
+
66
+
totalOps := 0
67
+
matchCount := 0
68
+
totalBytes := int64(0)
69
+
matchedBytes := int64(0)
70
+
71
+
startTime := time.Now()
72
+
73
+
for bundleNum := startBundle; bundleNum <= endBundle; bundleNum++ {
74
+
bundleFile := filepath.Join(bundleDir, fmt.Sprintf("%06d.jsonl.zst", bundleNum))
75
+
76
+
// Read compressed file
77
+
compressed, err := os.ReadFile(bundleFile)
78
+
if err != nil {
79
+
fmt.Fprintf(os.Stderr, "\nError reading bundle %d: %v\n", bundleNum, err)
80
+
continue
81
+
}
82
+
83
+
// Decompress with native gozstd (CGO, fast as native zstd)
84
+
decompressed, err := gozstd.Decompress(nil, compressed)
85
+
if err != nil {
86
+
fmt.Fprintf(os.Stderr, "\nError decompressing bundle %d: %v\n", bundleNum, err)
87
+
continue
88
+
}
89
+
90
+
// Split into lines
91
+
lines := bytes.Split(decompressed, []byte("\n"))
92
+
93
+
for position, line := range lines {
94
+
if len(line) == 0 {
95
+
continue
96
+
}
97
+
98
+
totalOps++
99
+
opSize := len(line)
100
+
totalBytes += int64(opSize)
101
+
102
+
// Parse JSON with go-json
103
+
var op Operation
104
+
if err := json.Unmarshal(line, &op); err != nil {
105
+
fmt.Fprintf(os.Stderr, "Error parsing operation: %v\n", err)
106
+
continue
107
+
}
108
+
109
+
// Run detect function
110
+
labels := detect(&op)
111
+
112
+
if len(labels) > 0 {
113
+
matchCount++
114
+
matchedBytes += int64(opSize)
115
+
116
+
// Extract last 4 chars of CID
117
+
cidShort := op.CID
118
+
if len(cidShort) > 4 {
119
+
cidShort = cidShort[len(cidShort)-4:]
120
+
}
121
+
122
+
// Write CSV line
123
+
writer.WriteString(strconv.Itoa(bundleNum))
124
+
writer.WriteByte(',')
125
+
writer.WriteString(strconv.Itoa(position))
126
+
writer.WriteByte(',')
127
+
writer.WriteString(cidShort)
128
+
writer.WriteByte(',')
129
+
writer.WriteString(strconv.Itoa(opSize))
130
+
writer.WriteByte(',')
131
+
writer.WriteString("0.95")
132
+
writer.WriteByte(',')
133
+
writer.WriteString(strings.Join(labels, ";"))
134
+
writer.WriteByte('\n')
135
+
}
136
+
}
137
+
138
+
// Progress
139
+
if bundleNum%10 == 0 {
140
+
elapsed := time.Since(startTime).Seconds()
141
+
opsPerSec := float64(totalOps) / elapsed
142
+
fmt.Fprintf(os.Stderr, "Processed %d/%d bundles | %d ops | %.0f ops/sec\r",
143
+
bundleNum, endBundle, totalOps, opsPerSec)
144
+
}
145
+
}
146
+
147
+
elapsed := time.Since(startTime).Seconds()
148
+
149
+
// Stats
150
+
fmt.Fprintf(os.Stderr, "\n\n")
151
+
fmt.Fprintf(os.Stderr, "✓ Detection complete\n")
152
+
fmt.Fprintf(os.Stderr, " Total operations: %d\n", totalOps)
153
+
fmt.Fprintf(os.Stderr, " Matches found: %d (%.2f%%)\n", matchCount, float64(matchCount)/float64(totalOps)*100)
154
+
fmt.Fprintf(os.Stderr, " Total size: %.1f MB\n", float64(totalBytes)/1e6)
155
+
fmt.Fprintf(os.Stderr, " Matched size: %.1f MB (%.2f%%)\n", float64(matchedBytes)/1e6, float64(matchedBytes)/float64(totalBytes)*100)
156
+
fmt.Fprintf(os.Stderr, "\n")
157
+
fmt.Fprintf(os.Stderr, " Time elapsed: %.2fs\n", elapsed)
158
+
fmt.Fprintf(os.Stderr, " Throughput: %.0f ops/sec\n", float64(totalOps)/elapsed)
159
+
fmt.Fprintf(os.Stderr, " Speed: %.1f MB/sec\n", float64(totalBytes)/elapsed/1e6)
160
+
}
+1
-1
scripts/detector-template.js
+1
-1
scripts/detector-template.js