Help us shape the future of Go! Take the 2026 Go Developer Survey

commands

package
v1.26.6 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 10, 2026 License: MIT Imports: 49 Imported by: 0

Documentation

Overview

Package commands provides functionality for the TDTP framework.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ActiveLicense 

func ActiveLicense() *license.License

ActiveLicense returns the resolved license, or the Community floor if ResolveLicense was never called (defensive default).

func BuildFixedFieldsForExport 

func BuildFixedFieldsForExport(schema packet.Schema, fixedFields []string) []string

BuildFixedFieldsForExport delegates to packet.ResolveFixedFields. Data-analysis fallback is not performed here (no rows yet at schema build time).

func CheckAdapter 

func CheckAdapter(lic *license.License, adapter string) error

CheckAdapter is GateAdapter against an explicit license; see CheckFeature.

func CheckFeature 

func CheckFeature(lic *license.License, feature string) error

CheckFeature is GateFeature against an explicit license rather than the process-wide one. The v2 CLI carries its license in Deps, so its gates do not depend on who called ResolveLicense last; the refusal text is the same for both CLIs. A nil license is the Community floor (fail closed).

func CheckPipelineVars 

func CheckPipelineVars(pkt *packet.DataPacket, expectVars map[string]string) error

CheckPipelineVars verifies PipelineContext variables in a packet against expected values. Returns error if a required variable is missing or has a different value. Called before any DB writes so import fails fast without side effects.

func ConvertTDTPToCSV 

func ConvertTDTPToCSV(ctx context.Context, opts CSVOptions) error

ConvertTDTPToCSV converts a TDTP packet to CSV.

Security gate:

  • v1.0 packets: pass-through, no security requirements.
  • v1.4 packets: VerifyAndPrepare (Mercury + local xxh3). MercuryURL set → full executor check, FallbackDegrade on timeout. MercuryURL empty → local integrity only (FallbackDegrade).
  • Any security failure → error returned, nothing written.

func ConvertTDTPToHTML 

func ConvertTDTPToHTML(ctx context.Context, opts HTMLOptions) error

ConvertTDTPToHTML converts a TDTP XML file to a beautiful standalone HTML page. Export is significantly faster than CSV because TDTP carries schema metadata, so the viewer renders correct column types without guessing.

Security gate:

  • v1.0 packets: pass-through, no security requirements.
  • v1.4 packets: VerifyAndPrepare (Mercury + local xxh3). MercuryURL set → full executor check, FallbackDegrade on timeout. MercuryURL empty → local integrity only (FallbackDegrade).
  • Any security failure → error returned, nothing written.

func ConvertTDTPToTDTP 

func ConvertTDTPToTDTP(ctx context.Context, opts ToTDTPOptions) error

ConvertTDTPToTDTP re-filters and/or re-versions an existing TDTP file into a new TDTP file, without a database round-trip. Shares its decompress/ security-gate/filter pipeline with ConvertTDTPToCSV — the differences are that the last step writes TDTP XML instead of CSV, and that encrypted input is refused rather than decrypted (see rejectEncryptedFile).

The explicit, mandatory version choice (--v1/--v13/--v14) is deliberate: a file produced by --to-tdtp always declares exactly which feature set it carries (integrity hashes, Dictionary) rather than leaving that ambiguous, the way a hand-assembled or third-party-generated file might.

func ConvertTDTPToXLSX 

func ConvertTDTPToXLSX(ctx context.Context, opts XLSXOptions) error

ConvertTDTPToXLSX converts a TDTP XML file to XLSX

func ConvertToCompact 

func ConvertToCompact(opts ConvertCompactOptions) error

ConvertToCompact reads a TDTP v1.x file and rewrites it in v1.3.1 compact format.

Auto-detection priority (applied to each packet independently):

  1. Explicit --fixed-fields list → mark those fields as fixed
  2. Fields with "_" prefix in name → strip prefix, mark fixed
  3. Data analysis → any column where all values are identical → mark fixed

If the input file is compressed, it is decompressed before processing and NOT re-compressed in the output (user can chain with --compress if needed).

func ConvertXLSXToTDTP 

func ConvertXLSXToTDTP(opts XLSXOptions) error

ConvertXLSXToTDTP converts an XLSX file to TDTP XML

func DecryptEncBlob 

func DecryptEncBlob(ctx context.Context, blob []byte, mercuryURL string) ([]byte, error)

DecryptEncBlob decrypts a binary blob produced by EncryptPacket / pipeline encryption. It extracts the package UUID from the blob header, retrieves the AES-256 key from xZMercury (burn-on-read), and returns the plaintext TDTP XML bytes.

mercuryURL must be non-empty; this function does not support dev-local mode because burn-on-read requires a running Mercury instance.

func DecryptEncFile 

func DecryptEncFile(ctx context.Context, path, mercuryURL string) ([]byte, error)

DecryptEncFile reads path, detects encryption, and returns plaintext TDTP XML. Non-encrypted files are returned as-is (pass-through).

func DecryptPacketV15 

func DecryptPacketV15(ctx context.Context, pkt *packet.DataPacket, mercuryURL string) error

DecryptPacketV15 retrieves the AES-256 key for pkt.Header.MessageID from xZMercury (burn-on-read — call once, reuse for every part of a multi-part message) and decrypts every encrypted section in place.

func DiffFiles 

func DiffFiles(ctx context.Context, options *DiffOptions) error

DiffFiles сравнивает два TDTP файла

func DiffFilesReport 

func DiffFilesReport(w io.Writer, ctx context.Context, options *DiffOptions) (*diff.DiffResult, error)

DiffFilesReport writes the v1 report and returns the comparison used to produce it, so callers can render structured output without recomputing.

func DiffFilesTo 

func DiffFilesTo(w io.Writer, ctx context.Context, options *DiffOptions) error

DiffFilesTo is DiffFiles writing its report to w instead of stdout, so embedders (tdtpcli_v2 --quiet/--json) control the output stream.

func EncryptPacket 

func EncryptPacket(ctx context.Context, pkt *packet.DataPacket, mercuryURL, pipelineName string) (blob []byte, packageUUID string, err error)

EncryptPacket serializes pkt to TDTP XML, then encrypts it via xZMercury BindKey. Returns the binary blob and the package UUID bound to the key.

pipelineName is embedded in the BindKey request for ACL enforcement. serverSecret is read from MERCURY_SERVER_SECRET env var when empty.

func EncryptPacketV15 

func EncryptPacketV15(ctx context.Context, pkt *packet.DataPacket, mercuryURL, pipelineName string) (xmlData []byte, packageUUID string, err error)

EncryptPacketV15 encrypts pkt in place using TDTP v1.5 section-level encryption (QueryContext/Schema/Data each turn opaque; Header stays plain XML) and returns the resulting XML bytes.

Unlike EncryptPacket (--enc13, legacy whole-blob format), the package UUID bound at xZMercury is always pkt.Header.MessageID — never a freshly generated one — because a v1.5 consumer must be able to read the UUID straight from the plain Header before decrypting anything. For multi-part packets, this is simply called once per part: each part already carries its own distinct Header.MessageID ("{base}-P{n}", set by GenerateReference), so each part's BindKey call lands on its own Redis key — no shared identifier, no overwrite race, no special handling needed (see docs/tdtp-protocol-schema.md → "v1.5" → "Multi-part packets").

pkt must already have ComputeIntegrity (and compression, if enabled) applied — this function does not run either; order is fixed (hash -> compress -> encrypt) and is the caller's responsibility.

func EncryptSectionsInPlace 

func EncryptSectionsInPlace(ctx context.Context, pkt *packet.DataPacket, mercuryURL, pipelineName string) (string, error)

EncryptSectionsInPlace — v1.5 шифрование БЕЗ сериализации: берёт ключ у xZMercury и шифрует секции пакета, оставляя его обычным *DataPacket.

Отделено от EncryptPacketV15 затем, чтобы шифрование стало шагом цепочки, а не поведением писателя. Пока оно жило в writePacket, оно было единственным преобразованием вне общего порядка — то есть тем местом, где ограничения из pkg/transform не проверялись вовсе.

Возвращает UUID пакета: он же ключ в Mercury, и вызывающий печатает его пользователю для последующей расшифровки.

func ExecutePipeline 

func ExecutePipeline(ctx context.Context, configPath string, opts PipelineOptions) error

ExecutePipeline executes an ETL pipeline from YAML configuration file.

Security levels:

  1. Code level: SQLValidator checks for forbidden keywords
  2. OS level: IsAdmin() checks administrator privileges for unsafe mode
  3. CLI level: This command enforces READ-ONLY by default
  4. SQL level: Only SELECT/WITH queries allowed in safe mode

Safe mode (default):

  • Only SELECT and WITH queries allowed
  • No data modification operations (INSERT, UPDATE, DELETE, DROP, etc.)
  • No admin privileges required

Unsafe mode (--unsafe flag, requires admin):

  • All SQL queries allowed
  • Administrator privileges required
  • Use with extreme caution

func ExportTable 

func ExportTable(ctx context.Context, config *adapters.Config, opts ExportOptions) error

ExportTable exports a table to TDTP XML file

func ExportTableToXLSX 

func ExportTableToXLSX(ctx context.Context, config *adapters.Config, opts XLSXOptions) error

ExportTableToXLSX exports a database table directly to XLSX

func ExportToBroker 

func ExportToBroker(ctx context.Context, dbConfig *adapters.Config, brokerCfg *BrokerConfig, tableName string, query *packet.Query, compress bool, compressLevel int, compressAlgo string, procMgr ProcessorManager, packetSizeMB int, mercuryURL string, encrypt, encryptLegacy bool) error

func ExportToBrokerWithOptions 

func ExportToBrokerWithOptions(ctx context.Context, dbConfig *adapters.Config, brokerCfg *BrokerConfig, tableName string, query *packet.Query, opts BrokerExportOptions) error

func GateAdapter 

func GateAdapter(adapter string) error

GateAdapter returns an error if the active license does not permit the adapter.

func GateFeature 

func GateFeature(feature string) error

GateFeature returns an error if the active license does not permit feature. Used to guard --enc (feature "enc"), --unsafe (feature "unsafe"), S3, etc.

func GateRowCount 

func GateRowCount(rowCount int) error

GateRowCount returns an error if rowCount exceeds the licensed per-export limit. A limit of 0 means unlimited.

func ImportFile 

func ImportFile(ctx context.Context, config *adapters.Config, opts ImportOptions) error

ImportFile imports a TDTP XML file (or multi-part set) to database. Parts are processed one at a time (streaming): each part is read, parsed, inserted, and released before the next part is loaded. This keeps memory usage constant regardless of the number of parts.

func ImportFromBroker 

func ImportFromBroker(ctx context.Context, dbConfig *adapters.Config, brokerCfg *BrokerConfig, opts ImportBrokerOptions) error

ImportFromBroker imports one complete export batch from the broker queue.

All packets of the same export share a common batch ID embedded in MessageID (e.g. "REF-20250319-ABCD1234-P1", "REF-20250319-ABCD1234-P2", ...). Packets from a different batch are Nack'd with requeue=true so they stay in the queue untouched. The function exits once all TotalParts are received or the idle timeout fires (queue empty).

func ImportXLSXToTable 

func ImportXLSXToTable(ctx context.Context, config *adapters.Config, opts XLSXOptions) error

ImportXLSXToTable imports an XLSX file directly to database table

func IncrementalSync 

func IncrementalSync(ctx context.Context, config *adapters.Config, opts SyncOptions) error

IncrementalSync performs incremental synchronization of a table

func InspectFile 

func InspectFile(ctx context.Context, inputFile string, storageCfg *storage.Config) error

InspectFile reads a TDTP XML file (local or s3://) and prints a clean YAML summary suitable for LLM/agent consumption. storageCfg may be nil for local files.

func InspectFileReport 

func InspectFileReport(w io.Writer, ctx context.Context, inputFile string, storageCfg *storage.Config) (*packet.DataPacket, error)

InspectFileReport writes the v1 report and returns the packet parsed for that report, so callers can render structured output without reading it again.

func InspectFileTo 

func InspectFileTo(w io.Writer, ctx context.Context, inputFile string, storageCfg *storage.Config) error

InspectFileTo is InspectFile writing its report to w instead of stdout, so embedders (tdtpcli_v2 --quiet/--json) control the output stream.

func InspectTable 

func InspectTable(ctx context.Context, config *adapters.Config, tableName string) error

InspectTable connects to the database and prints extended table metadata in YAML format suitable for agentic/LLM consumption.

tableName may include bracket-quoting: "[ZTR$Employee]" or "[dbo].[Orders]"

func InspectTableTo 

func InspectTableTo(w io.Writer, ctx context.Context, config *adapters.Config, tableName string) error

InspectTableTo is InspectTable writing its report to w instead of stdout, so embedders (tdtpcli_v2 --quiet/--json) control the output stream.

func IsCompressedFile 

func IsCompressedFile(filename string) bool

IsCompressedFile checks if filename suggests compressed content

func IsEncryptedBlob 

func IsEncryptedBlob(blob []byte) bool

IsEncryptedBlob reports whether blob carries the encryption header ([2B ver][1B algo][16B uuid]…). Content-based detection, independent of the file extension — a pipeline may write an encrypted blob to a .tdtp.xml path.

func IsEncryptedFile 

func IsEncryptedFile(path string) bool

IsEncryptedFile returns true when path ends with ".tdtp.enc" or ".enc". Used by import / converters to auto-detect encrypted input.

func IsEncryptedPacket 

func IsEncryptedPacket(pkt *packet.DataPacket) bool

IsEncryptedPacket reports whether a parsed DataPacket carries v1.5 section-level encryption (QueryContext/Schema/Data attribute-based), as opposed to IsEncryptedBlob's whole-packet binary envelope (v1.0-v1.4, --enc13). The two formats are mutually exclusive and auto-detected from different signals: IsEncryptedBlob inspects raw bytes before any XML parse; IsEncryptedPacket inspects an already-parsed packet's attributes.

func ListTables 

func ListTables(ctx context.Context, config *adapters.Config, pattern string) error

ListTables lists all tables in the database, optionally filtered by a glob pattern. pattern="" lists all tables; pattern="user*" lists only matching ones.

func ListTablesReport 

func ListTablesReport(w io.Writer, ctx context.Context, config *adapters.Config, pattern string) ([]string, error)

ListTablesReport writes the v1 report and returns the same filtered names for callers that also need a structured result.

func ListTablesTo 

func ListTablesTo(w io.Writer, ctx context.Context, config *adapters.Config, pattern string) error

ListTablesTo is ListTables writing its report to w instead of stdout, so embedders (tdtpcli_v2 --quiet/--json) control the output stream.

func ListViews 

func ListViews(ctx context.Context, config *adapters.Config) error

ListViews lists all database views with updatable status

func ListViewsReport 

func ListViewsReport(w io.Writer, ctx context.Context, config *adapters.Config) ([]string, error)

ListViewsReport writes the v1 report and returns the names from the same database query for callers that also need a structured result.

func ListViewsTo 

func ListViewsTo(w io.Writer, ctx context.Context, config *adapters.Config) error

ListViewsTo is ListViews writing its report to w instead of stdout, so embedders (tdtpcli_v2 --quiet/--json) control the output stream.

func ListenKafkaStream 

func ListenKafkaStream(ctx context.Context, dbConfig *adapters.Config, cfg ListenConfig) error

ListenKafkaStream runs the streaming consumer daemon. It blocks until SIGTERM/SIGINT is received or a fatal error occurs.

func MatchesPattern 

func MatchesPattern(name, pattern string) bool

matchesPattern reports whether name matches the given glob pattern. An empty pattern matches everything. Supports * and ? wildcards (path.Match semantics). % is treated as * for SQL-style patterns. Matching is case-insensitive.

func MergeFiles 

func MergeFiles(ctx context.Context, options MergeOptions) error

MergeFiles объединяет несколько TDTP файлов

func MergeFilesTo 

func MergeFilesTo(w io.Writer, ctx context.Context, options MergeOptions) error

MergeFilesTo is MergeFiles writing its report to w instead of stdout, so embedders (tdtpcli_v2 --quiet/--json) control the output stream.

func ParseImportStrategy 

func ParseImportStrategy(strategy string) (adapters.ImportStrategy, error)

ParseImportStrategy parses import strategy string

func PrintDiffHelp 

func PrintDiffHelp()

PrintDiffHelp выводит справку по команде diff

func PrintMergeHelp 

func PrintMergeHelp()

PrintMergeHelp выводит справку по команде merge

func ProcessRequest 

func ProcessRequest(ctx context.Context, opts ProcessRequestOptions) error

ProcessRequest читает TDTP request-пакет, выполняет запрос и генерирует response

func QuietOutput 

func QuietOutput() bool

QuietOutput reports whether --quiet is in effect.

func ResolveLicense 

func ResolveLicense(flagPath string) (*license.License, error)

ResolveLicense determines the license path, loads, and verifies it.

Path precedence:

  1. explicit --license flag (flagPath)
  2. TDTP_LICENSE environment variable
  3. ./tdtp.lic if it exists
  4. none → Community floor

Returns the active license. A present-but-invalid license is a hard error (fail closed): tampered or expired licenses must not silently downgrade.

func RunMap 

func RunMap(ctx context.Context, opts MapOptions) error

RunMap executes a cross-system field mapping: reads a TDTP packet, applies the field/enum remap from mappingFile, and upserts rows into the target DB. With opts.Listen=true it enters daemon mode, continuously consuming from the broker queue until SIGTERM/SIGINT.

func RunSteps 

func RunSteps(ctx context.Context, path string, vars map[string]string, quiet bool) error

RunSteps executes a multi-step workflow defined in a YAML file. Steps are run in dependency order (topological sort); steps with no ordering constraint between them run in parallel within each wave.

Workflow YAML format: 

name: my-workflow
description: Export → validate → map
steps:
  - id: export
    command: "--pipeline pipelines/export.yaml"
  - id: validate
    command: "--test out/export.tdtp.xml"
    depends_on: [export]
    on_error: skip
  - id: map
    command: "--map mappings/sync.yaml --input out/export.tdtp.xml"
    depends_on: [export]
    on_error: retry(3)

func RunStepsWithOptions 

func RunStepsWithOptions(ctx context.Context, path string, vars map[string]string, opts workflow.RunOptions) error

RunStepsWithOptions preserves the v1 workflow engine while allowing v2 to route progress and child output through its own output contract.

func SetQuietOutput 

func SetQuietOutput(q bool)

SetQuietOutput records the process-wide --quiet setting. Called once from main before dispatch.

func TestFile 

func TestFile(ctx context.Context, filePath string, storageCfg *storage.Config) error

TestFile performs a dry-run integrity check on a TDTP file or a multi-part batch. storageCfg may be nil for local files; required when filePath is an s3:// URI.

Checks performed:

  1. XML parse
  2. Missing parts (if multi-part batch detected)
  3. Consistent InReplyTo UUID and TableName across all parts
  4. Unique MessageID per part (no duplicates)
  5. RecordsInPart header vs actual row count
  6. XXH3 checksum (if present)
  7. Decompression (if compressed)

func TestFileTo 

func TestFileTo(w io.Writer, ctx context.Context, filePath string, storageCfg *storage.Config) error

TestFileTo is TestFile writing its report to w instead of stdout, so embedders (tdtpcli_v2 --quiet/--json) control the output stream.

func WriteErrorPacket 

func WriteErrorPacket(code, message, table, inReplyTo, serverMode, path string) error

WriteErrorPacket serializes a TDTP error packet to path. Used to produce a "receipt" when decryption fails so the pipeline receives a structured error instead of a missing output file. serverMode is the xZMercury server mode ("dev"/"prod") from the burn marker — empty string when not applicable. It populates AlarmDetails.ServerMode so that monitoring can filter dev-failover burns from prod-theft events. If path is empty or "-", writes to stdout.

Types

type BrokerConfig 

type BrokerConfig struct {
	Type           string
	Host           string
	Port           int
	User           string
	Password       string
	Queue          string
	VHost          string
	UseTLS         bool
	TLSSkipVerify  bool
	Exchange       string
	RoutingKey     string
	Durable        bool
	AutoDelete     bool
	Exclusive      bool
	PassiveDeclare bool
	QueuePath      string   // MSMQ: полный путь к очереди (например: ".\private$\tdtp_in")
	Brokers        []string // Kafka: список брокеров (["localhost:9092"])
	ConsumerGroup  string   // Kafka: consumer group ID
}

BrokerConfig holds broker configuration

func BrokerConfigFromCliconfig 

func BrokerConfigFromCliconfig(config *cliconfig.Config) BrokerConfig

BrokerConfigFromCliconfig builds the runtime broker config from the YAML model. The queue/topic comes exclusively from config, never from CLI flags: the operator owns the destination, the user only names the table. Shared by the v1 and v2 CLIs (v1's buildBrokerConfig delegates here).

type BrokerExportOptions 

type BrokerExportOptions struct {
	Compress      bool
	CompressLevel int
	CompressAlgo  string
	ProcessorMgr  ProcessorManager
	PacketSizeMB  int
	MercuryURL    string
	Encrypt       bool
	EncryptLegacy bool
	// IntegrityV14 stamps v1.4 xxh3 hashes before compression (local-only
	// without MercuryURL, registered with it) — the --integrity the file
	// export has and the broker path never did.
	IntegrityV14 bool
	// MercuryCaller overrides the sender identity on registration
	// (default: the table name). v1 accepts --mercury-caller here and
	// ignores it; v2 honors it, like it honors --translit where v1
	// does not — a flag must do what it says.
	MercuryCaller string
}

ExportToBroker exports table data to message broker.

mercuryURL non-empty + encrypt=true enables encryption (mirrors --enc's export.go behavior, previously unsupported here — confirmed by grep this command pair had zero encryption support before this):

  • encryptLegacy=false (default, v1.5): each packet's QueryContext/ Schema/Data go opaque via EncryptPacketV15, Header stays plain XML — the broker message is still valid XML, keyed by that packet's own Header.MessageID.
  • encryptLegacy=true (--enc13): whole-packet binary blob via EncryptPacket, same as --export --enc13 produces to a file.

BrokerExportOptions mirrors ExportToBroker's parameters, plus IntegrityV14. New code takes this; the positional form stays for v1 (frozen) and dies with it at wave 4 — fourteen positionals were already one too many (see the --quiet comment below), a fifteenth bool next to two others would be a step backwards from the thing it fixes.

type CSVOptions 

type CSVOptions struct {
	InputFile  string
	OutputFile string // "" or "-" → stdout
	Delimiter  rune   // field separator; default ','
	CP         string // code page: "utf8" (default), "1251", "866"
	BOM        bool   // prepend UTF-8 BOM (helps Excel auto-detect)
	Translit   bool   // transliterate non-ASCII field names to ASCII headers
	Query      *packet.Query

	// MercuryURL enables full executor verification for v1.4 packets.
	// Empty → local xxh3 integrity check only (FallbackDegrade policy).
	MercuryURL string
}

CSVOptions holds options for TDTP → CSV conversion.

type ConvertCompactOptions 

type ConvertCompactOptions struct {
	InputFile   string
	OutputFile  string
	FixedFields []string // explicit list; nil = auto-detect
	Tail        bool     // write tail row with all fixed fields explicit (stream validation / carry handoff)
	Query       *packet.Query
}

ConvertCompactOptions holds options for the --to-compact conversion command.

type DiffOptions 

type DiffOptions struct {
	FileA         string   // Первый TDTP файл
	FileB         string   // Второй TDTP файл
	KeyFields     []string // Ключевые поля (опционально)
	IgnoreFields  []string // Игнорировать поля
	CaseSensitive bool     // Учитывать регистр
	OutputFormat  string   // Формат вывода: text, json
}

DiffOptions опции для команды diff

type ExportOptions 

type ExportOptions struct {
	TableName        string
	OutputFile       string
	Query            *packet.Query
	Fields           []string // Column projection: nil/empty = all columns
	ProcessorMgr     ProcessorManager
	Compress         bool
	CompressLevel    int
	CompressAlgo     string // Алгоритм сжатия: "zstd" (по умолчанию) или "kanzi"
	EnableChecksum   bool   // Add XXH3 checksum for data integrity verification
	ReadOnlyFields   bool   // Include read-only fields (timestamp, computed, identity)
	Fast             bool   // Skip SpecialValues detection for maximum export speed
	Columnar         bool   // Write Data column-major (layout="columns")
	Stream           bool   // Потоковый экспорт (см. streamExportTable)
	PacketSizeMB     int    // --packet-size: размер одной части в МБ (0 = умолчание ~1.9 МБ)
	FallbackRowLimit int64  // Max rows for in-memory fallback when SQL pushdown fails (0 = unlimited)

	// v1.3.1 compact format
	Compact     bool     // Enable compact format output
	FixedFields []string // Explicit fixed field names; nil = auto-detect from _prefix
	CompactTail bool     // Write tail row with all fixed fields explicit

	// v1.4 integrity — xxh3_128 hashes (Schema + Data + Packet fingerprint).
	// Computed BEFORE compression so hashes cover plain-text rows.
	// Consumer must decompress first, then call pipeline.VerifyAndPrepare.
	IntegrityV14  bool   // Stamp packet with v1.4 xxh3_128 hashes
	MercuryURL    string // Optional: register hash in xzMercury (empty = local integrity only)
	MercuryCaller string // X-Caller header for Mercury registration (default: "tdtpcli")

	// Encryption (--enc / --enc13 tier). Requires MercuryURL.
	// Key is bound in xZMercury (burn-on-read).
	//
	// Encrypt=true, EncryptLegacy=false (--enc, default since v1.5):
	//   TDTP v1.5 section-level format — QueryContext/Schema/Data go opaque,
	//   Header stays plain XML. Key bound to pkt.Header.MessageID. Output
	//   stays a normal .tdtp.xml file (still valid XML).
	//   Consumer reads with: --import file.tdtp.xml --mercury-url http://...
	//
	// Encrypt=true, EncryptLegacy=true (--enc13):
	//   Legacy TDTP v1.3 whole-packet binary blob. Output file: <name>.tdtp.enc.
	//   Consumer reads with: --import file.tdtp.enc --mercury-url http://...
	Encrypt       bool // AES-256-GCM via xZMercury BindKey/RetrieveKey
	EncryptLegacy bool // true = --enc13 (whole-blob v1.3); false = --enc (v1.5 section-level, default)

	// Object storage (S3/SeaweedFS). Non-nil → stream to object storage instead of local file.
	StorageCfg *storage.Config // storage driver config with bucket
	StorageKey string          // object key within the bucket
}

ExportOptions holds options for export operations

type HTMLOptions 

type HTMLOptions struct {
	InputFile   string
	OutputFile  string        // if empty, auto-generated next to input file
	OpenBrowser bool          // open result in default browser
	Limit       int           // max rows to render (0 = all)
	RowStart    int           // first row to render, 1-indexed (0 = from beginning)
	RowEnd      int           // last row to render, 1-indexed inclusive (0 = to end)
	Query       *packet.Query // TDTQL query for WHERE/ORDER-BY/LIMIT/OFFSET filtering

	// MercuryURL enables full executor verification for v1.4 packets.
	// Empty → local xxh3 integrity check only (FallbackDegrade policy).
	MercuryURL string
}

HTMLOptions holds options for TDTP → HTML conversion

type ImportBrokerOptions 

type ImportBrokerOptions struct {
	Strategy    adapters.ImportStrategy
	TargetTable string // override table name from packet header (fixes name conflicts)
	OutputFile  string // if set, save packets to file(s) instead of importing to DB
	Raw         bool   // save raw bytes as-is, no parse/decompress/validate
	Keep        bool   // allow partial writes (non-atomic): import each part immediately.
	// Default (Keep=false): all parts committed in one transaction via ImportPackets — all-or-nothing.
	// Keep=true: each part committed individually; if a later part fails, earlier parts remain.
	// Use Keep for batches too large for a single DB transaction.
	IdleTimeout time.Duration // how long to wait for the next message before stopping (0 = default 5s)

	// PipelineContext precondition check (v1.4): --expect-var name=value.
	ExpectVars map[string]string

	// MercuryURL enables full executor verification for v1.4 packets.
	// Empty → local xxh3 integrity check only (FallbackDegrade policy).
	MercuryURL string
}

ImportBrokerOptions holds options for ImportFromBroker

type ImportOptions 

type ImportOptions struct {
	FilePath     string
	TargetTable  string   // Переопределяет имя таблицы из XML (опционально)
	Fields       []string // Column whitelist: only import these columns (nil/empty = all)
	Strategy     adapters.ImportStrategy
	ProcessorMgr ProcessorManager

	// Field name sanitization — applied to schema fields before import.
	// Row data is positional and is never modified.
	SanitizeClear    bool // replace special chars (%, @, #, space, …) with safe tokens
	SanitizeTranslit bool // transliterate non-ASCII chars to ASCII (Cyrillic, European)

	// Object storage (S3/SeaweedFS). Non-nil → download from object storage instead of local file.
	StorageCfg *storage.Config // storage driver config with bucket
	StorageKey string          // object key within the bucket

	// PipelineContext precondition check (v1.4): --expect-var name=value.
	// Import fails before any DB writes if packet variables don't match.
	ExpectVars map[string]string

	// MercuryURL enables full executor verification for v1.4 packets.
	// Empty → local xxh3 integrity check only (FallbackDegrade policy).
	MercuryURL string
}

ImportOptions holds options for import operations

type ListenConfig 

type ListenConfig struct {
	BrokerCfg  *BrokerConfig
	Strategy   adapters.ImportStrategy
	MercuryURL string    // v1.4 security gate; empty → local integrity only
	Output     io.Writer // nil preserves v1 stdout progress
}

ListenConfig holds configuration for the streaming consumer daemon.

type MapOptions 

type MapOptions struct {
	MappingFile string             // path to mapping.yaml
	InputFile   string             // path to source .tdtp.xml (or .tdtp.enc) file
	DryRun      bool               // print what would happen without writing to DB
	MercuryURL  string             // xZMercury base URL for decrypting .enc input (burn-on-read)
	Listen      bool               // daemon mode: loop on broker queue until SIGTERM
	Auditor     SyncAuditor        // per-message audit in loop mode; nil = disabled
	Output      io.Writer          // progress and results; nil keeps stdout for v1
	AdapterGate func(string) error // optional target-adapter license gate

	// Drain turns the loop into a unit of work: consume until the queue has
	// been idle this long, then exit. --listen never ends, which is right for
	// a daemon and wrong for anything that has to report a result — a
	// scheduled run, an orchestrator job, a CI step. One message per
	// invocation (the plain broker:// path) is the other extreme: it cannot
	// keep up with a burst.
	Drain time.Duration

	// Quiet reduces the run to its result: one line naming the target, the row
	// count and the elapsed time. Errors are never suppressed — a skipped
	// message still says why it was skipped, because that is the one thing a
	// scheduled run cannot afford to lose.
	Quiet bool
	// contains filtered or unexported fields
}

MapOptions holds parameters for the --map command.

type MergeOptions 

type MergeOptions struct {
	InputFiles    []string // TDTP файлы для объединения
	OutputFile    string   // Выходной файл
	Strategy      string   // Стратегия: union, intersection, left, right, append
	KeyFields     []string // Ключевые поля (опционально)
	Compress      bool     // Сжать результат
	ShowConflicts bool     // Показать конфликты
	// SortFields orders the merged rows by these columns for deterministic
	// output (union order follows Go map iteration and varies run to run).
	// Empty = keep engine order. Shared by v1/v2 CLIs; v2 exposes --sort.
	SortFields []string
	// SortDesc reverses the sort (v2 --order desc; default asc).
	SortDesc bool
}

MergeOptions опции для команды merge

type OpMetrics 

type OpMetrics struct {
	Resource        string
	RecordsAffected int64
}

OpMetrics carries the resource name and record count discovered deep inside a command's execution (export.go/import.go/pipeline.go) back up to main.go's audit-logging call site, without changing any command function's signature.

Why a context side channel instead of (Result, error) return values: tdtpcli has ~20 command branches (list, export, import, pipeline, xlsx, broker...) each calling a differently-shaped XxxOptions function. Threading a result struct through all of them for one cross-cutting concern (audit data) would touch every command function's public signature. context.Context is already the first parameter everywhere, so this is purely additive.

func WithOpMetrics 

func WithOpMetrics(ctx context.Context) (context.Context, *OpMetrics)

WithOpMetrics attaches a fresh OpMetrics to ctx and returns both — main.go keeps the pointer to read it back after the command finishes.

type PipelineOptions 

type PipelineOptions struct {
	Unsafe         bool              // --unsafe: разрешить все SQL операции (требует admin)
	UnsafeCertPath string            // --unsafe-cert: path to capability certificate
	Encrypt        bool              // --enc: переопределить output.tdtp.encryption: true (v1.5 section-level, по умолчанию с v1.5)
	EncryptLegacy  bool              // --enc13: legacy TDTP v1.3 whole-blob формат вместо v1.5
	EncDev         bool              // --enc-dev: использовать DevClient вместо xZMercury (только !production сборки)
	Variables      map[string]string // @name=value аргументы из CLI
}

PipelineOptions содержит опции выполнения ETL пайплайна.

type ProcessRequestOptions 

type ProcessRequestOptions struct {
	RequestFile   string             // Путь к входящему request.tdtp
	OutputFile    string             // Куда писать response (опционально, иначе авто)
	ConfigsDir    string             // Директория с конфигами вида {Recipient}.yaml
	DefaultConfig *adapters.Config   // Fallback если {Recipient}.yaml не найден
	CheckAdapter  func(string) error // optional license gate for recipient-specific configs
	Output        io.Writer          // nil preserves v1 stdout progress
}

ProcessRequestOptions holds options for process-request operation

type ProcessorManager 

type ProcessorManager interface {
	processors.PacketProcessor // Name() + ProcessPacket()
	HasProcessors() bool
}

ProcessorManager interface for applying data processors. Embeds processors.PacketProcessor so it can participate in PacketChain directly.

type RowProcessors 

type RowProcessors struct {
	// contains filtered or unexported fields
}

RowProcessors is the --mask/--validate/--normalize chain both CLIs build from flags (v1 through its ProcessorManager alias). It satisfies the ProcessorManager interface the engines take.

func NewRowProcessors 

func NewRowProcessors() *RowProcessors

NewRowProcessors creates a new processor manager

func (*RowProcessors) AddMaskProcessor 

func (pm *RowProcessors) AddMaskProcessor(maskFields string) error

AddMaskProcessor adds field masking processor from CLI flag Format: --mask email,phone,card

func (*RowProcessors) AddMaskRules 

func (pm *RowProcessors) AddMaskRules(rules []cliconfig.MaskRule) error

AddMaskRules builds a masker from config-file rules (cliconfig ProcessorsConfig.Mask): - mask: [{field: email, strategy: partial}]. A strategy naming a valid MaskPattern is honored; anything else falls back to name-based auto-detection — the same rule --mask applies to bare field names.

func (*RowProcessors) AddNormalizeProcessor 

func (pm *RowProcessors) AddNormalizeProcessor(rulesFile string) error

AddNormalizeProcessor adds field normalization processor from YAML file. Format: --normalize rules.yaml

YAML structure: 

fields:
  email: email
  phone: phone
  city: uppercase

Supported rules: email, phone, whitespace, uppercase, lowercase, date. A file without a "fields" section is an error, for the same reason as in AddValidateProcessor.

func (*RowProcessors) AddNormalizeRules 

func (pm *RowProcessors) AddNormalizeRules(rules []cliconfig.NormalizeRule) error

AddNormalizeRules builds a normalizer from config-file rules: - normalize: [{field: city, strategy: uppercase}]. Unknown strategies fail at build, like a bad rule file.

func (*RowProcessors) AddValidateProcessor 

func (pm *RowProcessors) AddValidateProcessor(rulesFile string) error

AddValidateProcessor adds field validation processor from YAML file. Format: --validate rules.yaml

YAML structure: 

rules:
  email: email
  age: range:0-150
  status: [required, "enum:active,inactive"]
on_error: fail          # fail (default) | filter | warn
stop_on_first_error: false   # optional, only for on_error: fail

on_error strategies:

  • fail — abort on errors, return full error list
  • filter — remove invalid rows, pass the rest (count printed to stderr)
  • warn — print warnings to stderr, pass all rows unchanged

A file without a "rules" section is an error. It used to skip silently: --validate naming the wrong file validated nothing and said so nowhere.

func (*RowProcessors) AddValidateRules 

func (pm *RowProcessors) AddValidateRules(rules []cliconfig.ValidateRule) error

AddValidateRules builds a validator from config-file rules: - validate: [{field: age, type: range, min: "0", max: "150"}]. Type must name a known rule (regex, range, enum, required, length, email, phone, url, date); the param is Pattern, else Min-Max joined exactly as the rule-file grammar spells it ("range:0-150"). Anything else is a usage error, like a bad rule file.

func (*RowProcessors) HasProcessors 

func (pm *RowProcessors) HasProcessors() bool

HasProcessors checks if any processors are configured

func (*RowProcessors) Name 

func (pm *RowProcessors) Name() string

Name implements processors.PacketProcessor.

func (*RowProcessors) ProcessPacket 

func (pm *RowProcessors) ProcessPacket(ctx context.Context, pkt *packet.DataPacket) error

ProcessPacket applies all processors to a packet's data

type SyncAuditor 

type SyncAuditor interface {
	RecordSync(ctx context.Context, resource string, records int64, duration time.Duration, err error)
}

SyncAuditor receives one audit record per message processed in daemon mode.

The audit entry is written when the process exits. For a one-shot --map that is the same as "after the work". A daemon runs until SIGTERM, so the same mechanism produced one record covering days of syncs: the run was audited, the work inside it was not.

Declared here rather than taking the logger directly because the audit machinery lives in package main, which imports this package.

nil disables it — that is what a one-shot run passes.

type SyncOptions 

type SyncOptions struct {
	TableName      string
	OutputFile     string
	TrackingField  string
	CheckpointFile string
	BatchSize      int
	Fields         []string // Column projection; tracking field is always included automatically
	ProcessorMgr   ProcessorManager

	// Quiet replaces the running commentary with one result line:
	// "<table>  <n> rows  <elapsed>". Everything it drops -- the table name,
	// the tracking field, the checkpoint path -- is already on the command
	// line that invoked this.
	Quiet  bool
	Output io.Writer // nil preserves v1 stdout progress
	Rows   *int64    // optional result count for structured CLI output

	// BrokerCfg sends the result to a queue instead of writing files.
	//
	// Incremental sync tracked a checkpoint but could only write to disk;
	// export could only send. Wanting both meant keeping the watermark
	// outside the tool and building the --where clause by hand.
	//
	// nil keeps the file output, which is what every existing caller gets.
	BrokerCfg *BrokerConfig

	// Broker send options, ignored unless BrokerCfg is set.
	Compress      bool
	CompressLevel int
	CompressAlgo  string
	Encrypt       bool
	EncryptLegacy bool
	MercuryURL    string
}

SyncOptions holds options for incremental sync operations

type ToTDTPOptions 

type ToTDTPOptions struct {
	InputFile  string
	OutputFile string
	Query      *packet.Query // --where/--order-by/--limit/--offset/--fields, same as --to-csv

	// Version is the target protocol version to write: "1.0", "1.3.1", or
	// "1.4" (default). See applyTargetVersion for what each strips/adds.
	Version string

	// MercuryURL enables full executor verification for v1.4 packets.
	// Empty → local xxh3 integrity check only (FallbackDegrade policy).
	MercuryURL string
}

ToTDTPOptions holds options for the --to-tdtp conversion command.

type XLSXOptions 

type XLSXOptions struct {
	InputFile    string
	OutputFile   string
	SheetName    string
	TableName    string
	Strategy     adapters.ImportStrategy
	Query        *packet.Query
	ProcessorMgr ProcessorManager
	// S3 output: non-nil → write to temp file, upload to S3, delete temp.
	StorageCfg *storage.Config
	StorageKey string // object key for output

	// Translit renders non-ASCII field names as ASCII headers.
	Translit bool

	// MercuryURL enables full executor verification for v1.4 packets.
	// Empty → local xxh3 integrity check only (FallbackDegrade policy).
	MercuryURL string
}

XLSXOptions holds options for XLSX operations

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.