Add complete chain adapters for all LUX native chains

- P-Chain: Validators, delegators, staking, networks (linear chain)
- X-Chain: Assets, UTXOs, transfers (DAG)
- A-Chain: AI attestations, models, inference, providers, receipts (DAG)
- B-Chain: Cross-chain bridge transfers and proofs (DAG)
- Q-Chain: Quantum finality proofs, stamps, Corona keys (DAG)
- T-Chain: MPC/Teleport sessions, threshold signatures (DAG)
- Z-Chain: ZK privacy, shielded transfers, nullifiers (DAG)

All adapters implement dag.Adapter or chain.Adapter interfaces with:
- Complete database schemas with proper indexes
- JSON-RPC integration using correct method prefixes
- Store/query methods for chain-specific data types
- Extended statistics tracking

Includes docker-compose.yml for full stack deployment.
This commit is contained in:
2025-12-12 00:07:56 +00:00
parent 6b98a16178
commit 84ec184f6a
10 changed files with 4347 additions and 48 deletions
+565
View File
@@ -0,0 +1,565 @@
// Copyright (c) 2025 Lux Partners Limited
// SPDX-License-Identifier: MIT
// Package achain provides the A-Chain (AI) adapter for the DAG indexer.
// Handles AI compute attestations, model hashes, and inference results.
package achain
import (
"bytes"
"context"
"database/sql"
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/luxfi/indexer/dag"
)
const (
// DefaultPort for A-Chain indexer API
DefaultPort = 4500
// DefaultDatabase name
DefaultDatabase = "explorer_achain"
)
// AttestationType categorizes AI attestations
type AttestationType string
const (
AttestationCompute AttestationType = "compute" // Compute job attestation
AttestationModel AttestationType = "model" // Model registration
AttestationInference AttestationType = "inference" // Inference result
AttestationTraining AttestationType = "training" // Training job
)
// ComputeAttestation represents an AI compute attestation
type ComputeAttestation struct {
ID string `json:"id"`
Type AttestationType `json:"type"`
ModelHash string `json:"modelHash,omitempty"`
InputHash string `json:"inputHash,omitempty"`
OutputHash string `json:"outputHash,omitempty"`
ComputeUnits uint64 `json:"computeUnits"`
Provider string `json:"provider"`
Requester string `json:"requester"`
Timestamp time.Time `json:"timestamp"`
Status string `json:"status"`
Metadata json.RawMessage `json:"metadata,omitempty"`
}
// ModelRegistration represents a registered AI model
type ModelRegistration struct {
ID string `json:"id"`
ModelHash string `json:"modelHash"`
Name string `json:"name"`
Version string `json:"version"`
Owner string `json:"owner"`
Framework string `json:"framework,omitempty"`
Parameters uint64 `json:"parameters,omitempty"`
Timestamp time.Time `json:"timestamp"`
Metadata json.RawMessage `json:"metadata,omitempty"`
}
// InferenceResult represents an AI inference result
type InferenceResult struct {
ID string `json:"id"`
ModelID string `json:"modelId"`
InputHash string `json:"inputHash"`
OutputHash string `json:"outputHash"`
Confidence float64 `json:"confidence,omitempty"`
Latency time.Duration `json:"latency"`
Provider string `json:"provider"`
Timestamp time.Time `json:"timestamp"`
Metadata json.RawMessage `json:"metadata,omitempty"`
}
// AIProvider represents an AI compute provider
type AIProvider struct {
ID string `json:"id"`
Address string `json:"address"`
Name string `json:"name"`
Reputation float64 `json:"reputation"`
Capacity uint64 `json:"capacity"`
Active bool `json:"active"`
Timestamp time.Time `json:"timestamp"`
Metadata json.RawMessage `json:"metadata,omitempty"`
}
// AIReceipt represents a payment receipt for AI compute services
type AIReceipt struct {
ID string `json:"id"`
AttestationID string `json:"attestationId"`
Provider string `json:"provider"`
Requester string `json:"requester"`
Amount uint64 `json:"amount"`
Currency string `json:"currency"`
Status string `json:"status"`
Timestamp time.Time `json:"timestamp"`
}
// TrainingJob represents an AI training job
type TrainingJob struct {
ID string `json:"id"`
ModelID string `json:"modelId"`
DatasetHash string `json:"datasetHash"`
Epochs int `json:"epochs"`
BatchSize int `json:"batchSize"`
LearningRate float64 `json:"learningRate"`
Provider string `json:"provider"`
Status string `json:"status"`
StartedAt time.Time `json:"startedAt,omitempty"`
CompletedAt time.Time `json:"completedAt,omitempty"`
Metadata json.RawMessage `json:"metadata,omitempty"`
}
// Adapter implements dag.Adapter for A-Chain
type Adapter struct {
rpcEndpoint string
httpClient *http.Client
}
// New creates a new A-Chain adapter
func New(rpcEndpoint string) *Adapter {
return &Adapter{
rpcEndpoint: rpcEndpoint,
httpClient: &http.Client{Timeout: 30 * time.Second},
}
}
// ParseVertex parses A-Chain vertex data from RPC response
func (a *Adapter) ParseVertex(data json.RawMessage) (*dag.Vertex, error) {
var raw struct {
ID string `json:"id"`
Type string `json:"type"`
ParentIDs []string `json:"parentIds"`
Height uint64 `json:"height"`
Epoch uint32 `json:"epoch"`
TxIDs []string `json:"txIds"`
Timestamp int64 `json:"timestamp"`
Status string `json:"status"`
Data json.RawMessage `json:"data"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, fmt.Errorf("unmarshal vertex: %w", err)
}
v := &dag.Vertex{
ID: raw.ID,
Type: raw.Type,
ParentIDs: raw.ParentIDs,
Height: raw.Height,
Epoch: raw.Epoch,
TxIDs: raw.TxIDs,
Status: dag.Status(raw.Status),
Data: raw.Data,
Metadata: make(map[string]interface{}),
}
if raw.Timestamp > 0 {
v.Timestamp = time.Unix(raw.Timestamp, 0)
} else {
v.Timestamp = time.Now()
}
// Parse AI-specific metadata from vertex data
if len(raw.Data) > 0 {
var aiData struct {
AttestationType string `json:"attestationType"`
ModelHash string `json:"modelHash"`
Provider string `json:"provider"`
ComputeUnits uint64 `json:"computeUnits"`
}
if err := json.Unmarshal(raw.Data, &aiData); err == nil {
if aiData.AttestationType != "" {
v.Metadata["attestation_type"] = aiData.AttestationType
}
if aiData.ModelHash != "" {
v.Metadata["model_hash"] = aiData.ModelHash
}
if aiData.Provider != "" {
v.Metadata["provider"] = aiData.Provider
}
if aiData.ComputeUnits > 0 {
v.Metadata["compute_units"] = aiData.ComputeUnits
}
}
}
return v, nil
}
// GetRecentVertices fetches recent vertices from A-Chain via RPC
func (a *Adapter) GetRecentVertices(ctx context.Context, limit int) ([]json.RawMessage, error) {
req := map[string]interface{}{
"jsonrpc": "2.0",
"id": 1,
"method": "avm.getRecentVertices",
"params": map[string]interface{}{
"limit": limit,
},
}
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("marshal request: %w", err)
}
httpReq, err := http.NewRequestWithContext(ctx, "POST", a.rpcEndpoint, bytes.NewReader(body))
if err != nil {
return nil, fmt.Errorf("create request: %w", err)
}
httpReq.Header.Set("Content-Type", "application/json")
resp, err := a.httpClient.Do(httpReq)
if err != nil {
return nil, fmt.Errorf("rpc call: %w", err)
}
defer resp.Body.Close()
var result struct {
Result struct {
Vertices []json.RawMessage `json:"vertices"`
} `json:"result"`
Error *struct {
Code int `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("decode response: %w", err)
}
if result.Error != nil {
return nil, fmt.Errorf("rpc error %d: %s", result.Error.Code, result.Error.Message)
}
return result.Result.Vertices, nil
}
// GetVertexByID fetches a specific vertex by ID
func (a *Adapter) GetVertexByID(ctx context.Context, id string) (json.RawMessage, error) {
req := map[string]interface{}{
"jsonrpc": "2.0",
"id": 1,
"method": "avm.getVertex",
"params": map[string]interface{}{
"vertexID": id,
},
}
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("marshal request: %w", err)
}
httpReq, err := http.NewRequestWithContext(ctx, "POST", a.rpcEndpoint, bytes.NewReader(body))
if err != nil {
return nil, fmt.Errorf("create request: %w", err)
}
httpReq.Header.Set("Content-Type", "application/json")
resp, err := a.httpClient.Do(httpReq)
if err != nil {
return nil, fmt.Errorf("rpc call: %w", err)
}
defer resp.Body.Close()
var result struct {
Result json.RawMessage `json:"result"`
Error *struct {
Code int `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("decode response: %w", err)
}
if result.Error != nil {
return nil, fmt.Errorf("rpc error %d: %s", result.Error.Code, result.Error.Message)
}
return result.Result, nil
}
// InitSchema creates A-Chain specific database tables
func (a *Adapter) InitSchema(db *sql.DB) error {
schema := `
-- AI Compute Attestations
CREATE TABLE IF NOT EXISTS achain_attestations (
id TEXT PRIMARY KEY,
type TEXT NOT NULL,
model_hash TEXT,
input_hash TEXT,
output_hash TEXT,
compute_units BIGINT DEFAULT 0,
provider TEXT,
requester TEXT,
status TEXT DEFAULT 'pending',
vertex_id TEXT,
timestamp TIMESTAMPTZ NOT NULL,
metadata JSONB,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_achain_attestations_type ON achain_attestations(type);
CREATE INDEX IF NOT EXISTS idx_achain_attestations_provider ON achain_attestations(provider);
CREATE INDEX IF NOT EXISTS idx_achain_attestations_model ON achain_attestations(model_hash);
CREATE INDEX IF NOT EXISTS idx_achain_attestations_timestamp ON achain_attestations(timestamp DESC);
-- AI Model Registry
CREATE TABLE IF NOT EXISTS achain_models (
id TEXT PRIMARY KEY,
model_hash TEXT UNIQUE NOT NULL,
name TEXT NOT NULL,
version TEXT,
owner TEXT NOT NULL,
framework TEXT,
parameters BIGINT DEFAULT 0,
vertex_id TEXT,
timestamp TIMESTAMPTZ NOT NULL,
metadata JSONB,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_achain_models_owner ON achain_models(owner);
CREATE INDEX IF NOT EXISTS idx_achain_models_hash ON achain_models(model_hash);
CREATE INDEX IF NOT EXISTS idx_achain_models_name ON achain_models(name);
-- AI Inference Results
CREATE TABLE IF NOT EXISTS achain_inferences (
id TEXT PRIMARY KEY,
model_id TEXT NOT NULL,
input_hash TEXT NOT NULL,
output_hash TEXT NOT NULL,
confidence NUMERIC(10,6),
latency_ms BIGINT,
provider TEXT,
vertex_id TEXT,
timestamp TIMESTAMPTZ NOT NULL,
metadata JSONB,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_achain_inferences_model ON achain_inferences(model_id);
CREATE INDEX IF NOT EXISTS idx_achain_inferences_provider ON achain_inferences(provider);
CREATE INDEX IF NOT EXISTS idx_achain_inferences_timestamp ON achain_inferences(timestamp DESC);
-- AI Training Jobs
CREATE TABLE IF NOT EXISTS achain_training (
id TEXT PRIMARY KEY,
model_id TEXT,
dataset_hash TEXT NOT NULL,
epochs INT,
batch_size INT,
learning_rate NUMERIC(10,8),
provider TEXT,
status TEXT DEFAULT 'pending',
vertex_id TEXT,
started_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ,
metadata JSONB,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_achain_training_model ON achain_training(model_id);
CREATE INDEX IF NOT EXISTS idx_achain_training_status ON achain_training(status);
CREATE INDEX IF NOT EXISTS idx_achain_training_provider ON achain_training(provider);
-- AI Compute Providers Registry
CREATE TABLE IF NOT EXISTS achain_providers (
id TEXT PRIMARY KEY,
address TEXT UNIQUE NOT NULL,
name TEXT,
reputation NUMERIC(5,2) DEFAULT 0,
capacity BIGINT DEFAULT 0,
active BOOLEAN DEFAULT true,
vertex_id TEXT,
timestamp TIMESTAMPTZ NOT NULL,
metadata JSONB,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_achain_providers_active ON achain_providers(active);
CREATE INDEX IF NOT EXISTS idx_achain_providers_address ON achain_providers(address);
-- AI Payment Receipts
CREATE TABLE IF NOT EXISTS achain_receipts (
id TEXT PRIMARY KEY,
attestation_id TEXT NOT NULL,
provider TEXT NOT NULL,
requester TEXT NOT NULL,
amount BIGINT NOT NULL,
currency TEXT DEFAULT 'AI',
status TEXT DEFAULT 'pending',
vertex_id TEXT,
timestamp TIMESTAMPTZ NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_achain_receipts_attestation ON achain_receipts(attestation_id);
CREATE INDEX IF NOT EXISTS idx_achain_receipts_provider ON achain_receipts(provider);
CREATE INDEX IF NOT EXISTS idx_achain_receipts_requester ON achain_receipts(requester);
-- Add missing index for requester queries
CREATE INDEX IF NOT EXISTS idx_achain_attestations_requester ON achain_attestations(requester);
-- A-Chain specific stats
CREATE TABLE IF NOT EXISTS achain_stats (
id INT PRIMARY KEY DEFAULT 1,
total_attestations BIGINT DEFAULT 0,
total_models BIGINT DEFAULT 0,
total_inferences BIGINT DEFAULT 0,
total_training_jobs BIGINT DEFAULT 0,
total_compute_units BIGINT DEFAULT 0,
unique_providers INT DEFAULT 0,
unique_requesters INT DEFAULT 0,
total_receipts BIGINT DEFAULT 0,
updated_at TIMESTAMPTZ DEFAULT NOW()
);
INSERT INTO achain_stats (id) VALUES (1) ON CONFLICT DO NOTHING;
`
if _, err := db.Exec(schema); err != nil {
return fmt.Errorf("init achain schema: %w", err)
}
return nil
}
// GetStats returns A-Chain specific statistics
func (a *Adapter) GetStats(ctx context.Context, db *sql.DB) (map[string]interface{}, error) {
stats := make(map[string]interface{})
// Update stats first
_, _ = db.ExecContext(ctx, `
UPDATE achain_stats SET
total_attestations = (SELECT COUNT(*) FROM achain_attestations),
total_models = (SELECT COUNT(*) FROM achain_models),
total_inferences = (SELECT COUNT(*) FROM achain_inferences),
total_compute_units = (SELECT COALESCE(SUM(compute_units), 0) FROM achain_attestations),
unique_providers = (SELECT COUNT(DISTINCT provider) FROM achain_attestations WHERE provider IS NOT NULL),
unique_requesters = (SELECT COUNT(DISTINCT requester) FROM achain_attestations WHERE requester IS NOT NULL),
updated_at = NOW()
WHERE id = 1
`)
row := db.QueryRowContext(ctx, `
SELECT total_attestations, total_models, total_inferences,
total_compute_units, unique_providers, unique_requesters
FROM achain_stats WHERE id = 1
`)
var totalAttestations, totalModels, totalInferences, totalComputeUnits int64
var uniqueProviders, uniqueRequesters int
if err := row.Scan(&totalAttestations, &totalModels, &totalInferences,
&totalComputeUnits, &uniqueProviders, &uniqueRequesters); err != nil {
return nil, fmt.Errorf("scan stats: %w", err)
}
stats["total_attestations"] = totalAttestations
stats["total_models"] = totalModels
stats["total_inferences"] = totalInferences
stats["total_compute_units"] = totalComputeUnits
stats["unique_providers"] = uniqueProviders
stats["unique_requesters"] = uniqueRequesters
// Get attestation breakdown by type
rows, err := db.QueryContext(ctx, `
SELECT type, COUNT(*) FROM achain_attestations GROUP BY type
`)
if err == nil {
defer rows.Close()
breakdown := make(map[string]int64)
for rows.Next() {
var t string
var c int64
if rows.Scan(&t, &c) == nil {
breakdown[t] = c
}
}
stats["attestations_by_type"] = breakdown
}
return stats, nil
}
// StoreAttestation stores a compute attestation
func (a *Adapter) StoreAttestation(ctx context.Context, db *sql.DB, att *ComputeAttestation, vertexID string) error {
_, err := db.ExecContext(ctx, `
INSERT INTO achain_attestations (id, type, model_hash, input_hash, output_hash,
compute_units, provider, requester, status, vertex_id, timestamp, metadata)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status
`, att.ID, att.Type, att.ModelHash, att.InputHash, att.OutputHash,
att.ComputeUnits, att.Provider, att.Requester, att.Status, vertexID, att.Timestamp, att.Metadata)
return err
}
// StoreModel stores a model registration
func (a *Adapter) StoreModel(ctx context.Context, db *sql.DB, model *ModelRegistration, vertexID string) error {
_, err := db.ExecContext(ctx, `
INSERT INTO achain_models (id, model_hash, name, version, owner,
framework, parameters, vertex_id, timestamp, metadata)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
ON CONFLICT (id) DO NOTHING
`, model.ID, model.ModelHash, model.Name, model.Version, model.Owner,
model.Framework, model.Parameters, vertexID, model.Timestamp, model.Metadata)
return err
}
// StoreInference stores an inference result
func (a *Adapter) StoreInference(ctx context.Context, db *sql.DB, inf *InferenceResult, vertexID string) error {
_, err := db.ExecContext(ctx, `
INSERT INTO achain_inferences (id, model_id, input_hash, output_hash,
confidence, latency_ms, provider, vertex_id, timestamp, metadata)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
ON CONFLICT (id) DO NOTHING
`, inf.ID, inf.ModelID, inf.InputHash, inf.OutputHash,
inf.Confidence, inf.Latency.Milliseconds(), inf.Provider, vertexID, inf.Timestamp, inf.Metadata)
return err
}
// StoreTrainingJob stores a training job
func (a *Adapter) StoreTrainingJob(ctx context.Context, db *sql.DB, job *TrainingJob, vertexID string) error {
_, err := db.ExecContext(ctx, `
INSERT INTO achain_training (id, model_id, dataset_hash, epochs, batch_size,
learning_rate, provider, status, vertex_id, started_at, completed_at, metadata)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
ON CONFLICT (id) DO UPDATE SET
status = EXCLUDED.status,
completed_at = EXCLUDED.completed_at
`, job.ID, job.ModelID, job.DatasetHash, job.Epochs, job.BatchSize,
job.LearningRate, job.Provider, job.Status, vertexID, job.StartedAt, job.CompletedAt, job.Metadata)
return err
}
// StoreProvider stores or updates a provider registration
func (a *Adapter) StoreProvider(ctx context.Context, db *sql.DB, provider *AIProvider, vertexID string) error {
_, err := db.ExecContext(ctx, `
INSERT INTO achain_providers (id, address, name, reputation, capacity,
active, vertex_id, timestamp, metadata)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (id) DO UPDATE SET
reputation = EXCLUDED.reputation,
capacity = EXCLUDED.capacity,
active = EXCLUDED.active
`, provider.ID, provider.Address, provider.Name, provider.Reputation, provider.Capacity,
provider.Active, vertexID, provider.Timestamp, provider.Metadata)
return err
}
// StoreReceipt stores a payment receipt
func (a *Adapter) StoreReceipt(ctx context.Context, db *sql.DB, receipt *AIReceipt, vertexID string) error {
_, err := db.ExecContext(ctx, `
INSERT INTO achain_receipts (id, attestation_id, provider, requester,
amount, currency, status, vertex_id, timestamp)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status
`, receipt.ID, receipt.AttestationID, receipt.Provider, receipt.Requester,
receipt.Amount, receipt.Currency, receipt.Status, vertexID, receipt.Timestamp)
return err
}
// Verify interface compliance at compile time
var _ dag.Adapter = (*Adapter)(nil)
+517
View File
@@ -0,0 +1,517 @@
// Copyright (c) 2025 Lux Partners Limited
// SPDX-License-Identifier: MIT
// Package bchain provides the B-Chain (Bridge) adapter for cross-chain bridge operations.
// B-Chain uses DAG consensus for fast finality of bridge transfers.
package bchain
import (
"bytes"
"context"
"database/sql"
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/luxfi/indexer/dag"
)
// Default configuration
const (
DefaultPort = 4600
DefaultDatabase = "explorer_bchain"
RPCMethod = "bvm"
)
// BridgeStatus indicates the state of a cross-chain transfer
type BridgeStatus string
const (
BridgeStatusPending BridgeStatus = "pending"
BridgeStatusLocked BridgeStatus = "locked"
BridgeStatusConfirmed BridgeStatus = "confirmed"
BridgeStatusReleased BridgeStatus = "released"
BridgeStatusFailed BridgeStatus = "failed"
)
// Transfer represents a cross-chain bridge transfer
type Transfer struct {
ID string `json:"id"`
SourceChain string `json:"sourceChain"`
DestChain string `json:"destChain"`
SourceTxID string `json:"sourceTxId"`
DestTxID string `json:"destTxId,omitempty"`
Sender string `json:"sender"`
Recipient string `json:"recipient"`
AssetID string `json:"assetId"`
Amount string `json:"amount"`
Fee string `json:"fee,omitempty"`
Status BridgeStatus `json:"status"`
LockHeight uint64 `json:"lockHeight,omitempty"`
ReleaseHeight uint64 `json:"releaseHeight,omitempty"`
ProofHash string `json:"proofHash,omitempty"`
Timestamp time.Time `json:"timestamp"`
}
// Proof represents a bridge proof for cross-chain validation
type Proof struct {
ID string `json:"id"`
TransferID string `json:"transferId"`
ProofType string `json:"proofType"` // merkle, signature, zk
ProofData string `json:"proofData"`
Validators []string `json:"validators,omitempty"`
Signatures []string `json:"signatures,omitempty"`
Verified bool `json:"verified"`
VerifiedAt time.Time `json:"verifiedAt,omitempty"`
CreatedAt time.Time `json:"createdAt"`
}
// LockedAsset represents an asset locked in the bridge
type LockedAsset struct {
AssetID string `json:"assetId"`
SourceChain string `json:"sourceChain"`
Amount string `json:"amount"`
LockTxID string `json:"lockTxId"`
LockedAt time.Time `json:"lockedAt"`
UnlockedAt time.Time `json:"unlockedAt,omitempty"`
}
// vertexData holds B-Chain specific vertex data
type vertexData struct {
VertexID string `json:"vertexId"`
Type string `json:"type"`
ParentIDs []string `json:"parentIds"`
Height uint64 `json:"height"`
Epoch uint32 `json:"epoch"`
Timestamp int64 `json:"timestamp"`
Status string `json:"status"`
Transfers []Transfer `json:"transfers,omitempty"`
Proofs []Proof `json:"proofs,omitempty"`
LockedAssets []LockedAsset `json:"lockedAssets,omitempty"`
}
// Adapter implements dag.Adapter for B-Chain bridge operations
type Adapter struct {
rpcEndpoint string
httpClient *http.Client
}
// New creates a new B-Chain adapter
func New(rpcEndpoint string) *Adapter {
return &Adapter{
rpcEndpoint: rpcEndpoint,
httpClient: &http.Client{Timeout: 30 * time.Second},
}
}
// ParseVertex parses B-Chain vertex data from RPC response
func (a *Adapter) ParseVertex(data json.RawMessage) (*dag.Vertex, error) {
var vd vertexData
if err := json.Unmarshal(data, &vd); err != nil {
return nil, fmt.Errorf("parse vertex: %w", err)
}
// Extract transfer count for metadata
metadata := map[string]interface{}{
"transferCount": len(vd.Transfers),
"proofCount": len(vd.Proofs),
"lockedAssetCount": len(vd.LockedAssets),
}
// Collect unique chains involved
chains := make(map[string]bool)
for _, t := range vd.Transfers {
chains[t.SourceChain] = true
chains[t.DestChain] = true
}
chainList := make([]string, 0, len(chains))
for c := range chains {
chainList = append(chainList, c)
}
metadata["chainsInvolved"] = chainList
return &dag.Vertex{
ID: vd.VertexID,
Type: vd.Type,
ParentIDs: vd.ParentIDs,
Height: vd.Height,
Epoch: vd.Epoch,
Timestamp: time.Unix(vd.Timestamp, 0),
Status: dag.Status(vd.Status),
Data: data,
Metadata: metadata,
}, nil
}
// GetRecentVertices fetches recent vertices from the B-Chain RPC
func (a *Adapter) GetRecentVertices(ctx context.Context, limit int) ([]json.RawMessage, error) {
req := map[string]interface{}{
"jsonrpc": "2.0",
"id": 1,
"method": "bvm.getRecentVertices",
"params": map[string]interface{}{
"limit": limit,
},
}
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("marshal request: %w", err)
}
httpReq, err := http.NewRequestWithContext(ctx, "POST", a.rpcEndpoint, bytes.NewReader(body))
if err != nil {
return nil, fmt.Errorf("create request: %w", err)
}
httpReq.Header.Set("Content-Type", "application/json")
resp, err := a.httpClient.Do(httpReq)
if err != nil {
return nil, fmt.Errorf("rpc call: %w", err)
}
defer resp.Body.Close()
var result struct {
Result struct {
Vertices []json.RawMessage `json:"vertices"`
} `json:"result"`
Error *struct {
Code int `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("decode response: %w", err)
}
if result.Error != nil {
return nil, fmt.Errorf("rpc error %d: %s", result.Error.Code, result.Error.Message)
}
return result.Result.Vertices, nil
}
// GetVertexByID fetches a specific vertex by ID
func (a *Adapter) GetVertexByID(ctx context.Context, id string) (json.RawMessage, error) {
req := map[string]interface{}{
"jsonrpc": "2.0",
"id": 1,
"method": "bvm.getVertex",
"params": map[string]interface{}{
"vertexId": id,
},
}
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("marshal request: %w", err)
}
httpReq, err := http.NewRequestWithContext(ctx, "POST", a.rpcEndpoint, bytes.NewReader(body))
if err != nil {
return nil, fmt.Errorf("create request: %w", err)
}
httpReq.Header.Set("Content-Type", "application/json")
resp, err := a.httpClient.Do(httpReq)
if err != nil {
return nil, fmt.Errorf("rpc call: %w", err)
}
defer resp.Body.Close()
var result struct {
Result json.RawMessage `json:"result"`
Error *struct {
Code int `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("decode response: %w", err)
}
if result.Error != nil {
return nil, fmt.Errorf("rpc error %d: %s", result.Error.Code, result.Error.Message)
}
return result.Result, nil
}
// InitSchema creates B-Chain specific database tables
func (a *Adapter) InitSchema(db *sql.DB) error {
schema := `
-- Bridge transfers table
CREATE TABLE IF NOT EXISTS bchain_transfers (
id TEXT PRIMARY KEY,
vertex_id TEXT NOT NULL,
source_chain TEXT NOT NULL,
dest_chain TEXT NOT NULL,
source_tx_id TEXT,
dest_tx_id TEXT,
sender TEXT NOT NULL,
recipient TEXT NOT NULL,
asset_id TEXT NOT NULL,
amount TEXT NOT NULL,
fee TEXT,
status TEXT DEFAULT 'pending',
lock_height BIGINT,
release_height BIGINT,
proof_hash TEXT,
timestamp TIMESTAMPTZ NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_bchain_transfers_vertex ON bchain_transfers(vertex_id);
CREATE INDEX IF NOT EXISTS idx_bchain_transfers_source ON bchain_transfers(source_chain);
CREATE INDEX IF NOT EXISTS idx_bchain_transfers_dest ON bchain_transfers(dest_chain);
CREATE INDEX IF NOT EXISTS idx_bchain_transfers_sender ON bchain_transfers(sender);
CREATE INDEX IF NOT EXISTS idx_bchain_transfers_recipient ON bchain_transfers(recipient);
CREATE INDEX IF NOT EXISTS idx_bchain_transfers_asset ON bchain_transfers(asset_id);
CREATE INDEX IF NOT EXISTS idx_bchain_transfers_status ON bchain_transfers(status);
CREATE INDEX IF NOT EXISTS idx_bchain_transfers_timestamp ON bchain_transfers(timestamp DESC);
-- Bridge proofs table
CREATE TABLE IF NOT EXISTS bchain_proofs (
id TEXT PRIMARY KEY,
transfer_id TEXT NOT NULL REFERENCES bchain_transfers(id),
proof_type TEXT NOT NULL,
proof_data TEXT NOT NULL,
validators JSONB DEFAULT '[]',
signatures JSONB DEFAULT '[]',
verified BOOLEAN DEFAULT FALSE,
verified_at TIMESTAMPTZ,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_bchain_proofs_transfer ON bchain_proofs(transfer_id);
CREATE INDEX IF NOT EXISTS idx_bchain_proofs_type ON bchain_proofs(proof_type);
CREATE INDEX IF NOT EXISTS idx_bchain_proofs_verified ON bchain_proofs(verified);
-- Locked assets table
CREATE TABLE IF NOT EXISTS bchain_locked_assets (
id SERIAL PRIMARY KEY,
asset_id TEXT NOT NULL,
source_chain TEXT NOT NULL,
amount TEXT NOT NULL,
lock_tx_id TEXT NOT NULL,
locked_at TIMESTAMPTZ NOT NULL,
unlocked_at TIMESTAMPTZ,
UNIQUE(asset_id, source_chain, lock_tx_id)
);
CREATE INDEX IF NOT EXISTS idx_bchain_locked_asset ON bchain_locked_assets(asset_id);
CREATE INDEX IF NOT EXISTS idx_bchain_locked_chain ON bchain_locked_assets(source_chain);
-- Chain statistics table
CREATE TABLE IF NOT EXISTS bchain_chain_stats (
chain_id TEXT PRIMARY KEY,
total_transfers_in BIGINT DEFAULT 0,
total_transfers_out BIGINT DEFAULT 0,
total_volume_locked TEXT DEFAULT '0',
active_transfers BIGINT DEFAULT 0,
updated_at TIMESTAMPTZ DEFAULT NOW()
);
-- Bridge statistics table
CREATE TABLE IF NOT EXISTS bchain_bridge_stats (
id INT PRIMARY KEY DEFAULT 1,
total_transfers BIGINT DEFAULT 0,
pending_transfers BIGINT DEFAULT 0,
completed_transfers BIGINT DEFAULT 0,
failed_transfers BIGINT DEFAULT 0,
total_proofs BIGINT DEFAULT 0,
verified_proofs BIGINT DEFAULT 0,
unique_assets BIGINT DEFAULT 0,
unique_chains BIGINT DEFAULT 0,
updated_at TIMESTAMPTZ DEFAULT NOW()
);
INSERT INTO bchain_bridge_stats (id) VALUES (1) ON CONFLICT DO NOTHING;
`
if _, err := db.Exec(schema); err != nil {
return fmt.Errorf("init bchain schema: %w", err)
}
return nil
}
// GetStats returns B-Chain specific statistics
func (a *Adapter) GetStats(ctx context.Context, db *sql.DB) (map[string]interface{}, error) {
stats := make(map[string]interface{})
// Bridge stats
var bridgeStats struct {
TotalTransfers int64
PendingTransfers int64
CompletedTransfers int64
FailedTransfers int64
TotalProofs int64
VerifiedProofs int64
UniqueAssets int64
UniqueChains int64
}
err := db.QueryRowContext(ctx, `
SELECT total_transfers, pending_transfers, completed_transfers, failed_transfers,
total_proofs, verified_proofs, unique_assets, unique_chains
FROM bchain_bridge_stats WHERE id = 1
`).Scan(
&bridgeStats.TotalTransfers,
&bridgeStats.PendingTransfers,
&bridgeStats.CompletedTransfers,
&bridgeStats.FailedTransfers,
&bridgeStats.TotalProofs,
&bridgeStats.VerifiedProofs,
&bridgeStats.UniqueAssets,
&bridgeStats.UniqueChains,
)
if err != nil && err != sql.ErrNoRows {
return nil, fmt.Errorf("get bridge stats: %w", err)
}
stats["bridge"] = map[string]interface{}{
"total_transfers": bridgeStats.TotalTransfers,
"pending_transfers": bridgeStats.PendingTransfers,
"completed_transfers": bridgeStats.CompletedTransfers,
"failed_transfers": bridgeStats.FailedTransfers,
"total_proofs": bridgeStats.TotalProofs,
"verified_proofs": bridgeStats.VerifiedProofs,
"unique_assets": bridgeStats.UniqueAssets,
"unique_chains": bridgeStats.UniqueChains,
}
// Chain-specific stats
rows, err := db.QueryContext(ctx, `
SELECT chain_id, total_transfers_in, total_transfers_out, total_volume_locked, active_transfers
FROM bchain_chain_stats
`)
if err != nil {
return stats, nil // Return partial stats on error
}
defer rows.Close()
chainStats := make(map[string]interface{})
for rows.Next() {
var chainID string
var in, out, active int64
var volume string
if err := rows.Scan(&chainID, &in, &out, &volume, &active); err != nil {
continue
}
chainStats[chainID] = map[string]interface{}{
"transfers_in": in,
"transfers_out": out,
"volume_locked": volume,
"active_transfers": active,
}
}
stats["chains"] = chainStats
return stats, nil
}
// StoreTransfer stores a bridge transfer
func (a *Adapter) StoreTransfer(ctx context.Context, db *sql.DB, vertexID string, t *Transfer) error {
_, err := db.ExecContext(ctx, `
INSERT INTO bchain_transfers (
id, vertex_id, source_chain, dest_chain, source_tx_id, dest_tx_id,
sender, recipient, asset_id, amount, fee, status,
lock_height, release_height, proof_hash, timestamp
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
ON CONFLICT (id) DO UPDATE SET
status = EXCLUDED.status,
dest_tx_id = EXCLUDED.dest_tx_id,
release_height = EXCLUDED.release_height,
proof_hash = EXCLUDED.proof_hash
`,
t.ID, vertexID, t.SourceChain, t.DestChain, t.SourceTxID, t.DestTxID,
t.Sender, t.Recipient, t.AssetID, t.Amount, t.Fee, t.Status,
t.LockHeight, t.ReleaseHeight, t.ProofHash, t.Timestamp,
)
return err
}
// StoreProof stores a bridge proof
func (a *Adapter) StoreProof(ctx context.Context, db *sql.DB, p *Proof) error {
validators, _ := json.Marshal(p.Validators)
signatures, _ := json.Marshal(p.Signatures)
_, err := db.ExecContext(ctx, `
INSERT INTO bchain_proofs (
id, transfer_id, proof_type, proof_data, validators, signatures, verified, verified_at, created_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (id) DO UPDATE SET
verified = EXCLUDED.verified,
verified_at = EXCLUDED.verified_at
`,
p.ID, p.TransferID, p.ProofType, p.ProofData, validators, signatures, p.Verified, p.VerifiedAt, p.CreatedAt,
)
return err
}
// UpdateBridgeStats updates aggregate bridge statistics
func (a *Adapter) UpdateBridgeStats(ctx context.Context, db *sql.DB) error {
_, err := db.ExecContext(ctx, `
UPDATE bchain_bridge_stats SET
total_transfers = (SELECT COUNT(*) FROM bchain_transfers),
pending_transfers = (SELECT COUNT(*) FROM bchain_transfers WHERE status = 'pending' OR status = 'locked'),
completed_transfers = (SELECT COUNT(*) FROM bchain_transfers WHERE status = 'released'),
failed_transfers = (SELECT COUNT(*) FROM bchain_transfers WHERE status = 'failed'),
total_proofs = (SELECT COUNT(*) FROM bchain_proofs),
verified_proofs = (SELECT COUNT(*) FROM bchain_proofs WHERE verified = TRUE),
unique_assets = (SELECT COUNT(DISTINCT asset_id) FROM bchain_transfers),
unique_chains = (
SELECT COUNT(DISTINCT chain) FROM (
SELECT source_chain AS chain FROM bchain_transfers
UNION
SELECT dest_chain AS chain FROM bchain_transfers
) chains
),
updated_at = NOW()
WHERE id = 1
`)
return err
}
// GetTransfersByChain retrieves transfers for a specific chain
func (a *Adapter) GetTransfersByChain(ctx context.Context, db *sql.DB, chainID string, direction string, limit int) ([]Transfer, error) {
var query string
switch direction {
case "in":
query = `SELECT id, source_chain, dest_chain, source_tx_id, dest_tx_id, sender, recipient,
asset_id, amount, fee, status, lock_height, release_height, proof_hash, timestamp
FROM bchain_transfers WHERE dest_chain = $1 ORDER BY timestamp DESC LIMIT $2`
case "out":
query = `SELECT id, source_chain, dest_chain, source_tx_id, dest_tx_id, sender, recipient,
asset_id, amount, fee, status, lock_height, release_height, proof_hash, timestamp
FROM bchain_transfers WHERE source_chain = $1 ORDER BY timestamp DESC LIMIT $2`
default:
query = `SELECT id, source_chain, dest_chain, source_tx_id, dest_tx_id, sender, recipient,
asset_id, amount, fee, status, lock_height, release_height, proof_hash, timestamp
FROM bchain_transfers WHERE source_chain = $1 OR dest_chain = $1 ORDER BY timestamp DESC LIMIT $2`
}
rows, err := db.QueryContext(ctx, query, chainID, limit)
if err != nil {
return nil, fmt.Errorf("query transfers: %w", err)
}
defer rows.Close()
var transfers []Transfer
for rows.Next() {
var t Transfer
if err := rows.Scan(
&t.ID, &t.SourceChain, &t.DestChain, &t.SourceTxID, &t.DestTxID,
&t.Sender, &t.Recipient, &t.AssetID, &t.Amount, &t.Fee,
&t.Status, &t.LockHeight, &t.ReleaseHeight, &t.ProofHash, &t.Timestamp,
); err != nil {
continue
}
transfers = append(transfers, t)
}
return transfers, nil
}
// Verify interface compliance at compile time
var _ dag.Adapter = (*Adapter)(nil)
+223 -48
View File
@@ -16,8 +16,15 @@ import (
_ "github.com/lib/pq"
"github.com/luxfi/indexer/achain"
"github.com/luxfi/indexer/bchain"
"github.com/luxfi/indexer/chain"
"github.com/luxfi/indexer/dag"
"github.com/luxfi/indexer/pchain"
"github.com/luxfi/indexer/qchain"
"github.com/luxfi/indexer/tchain"
"github.com/luxfi/indexer/xchain"
"github.com/luxfi/indexer/zchain"
)
var version = "dev"
@@ -27,9 +34,10 @@ func main() {
chainType = flag.String("chain", "", "Chain to index: xchain, achain, bchain, qchain, tchain, zchain (DAG) or pchain (linear)")
rpcEndpoint = flag.String("rpc", "", "RPC endpoint (e.g., http://localhost:9630/ext/bc/X)")
databaseURL = flag.String("db", "", "PostgreSQL connection URL")
httpPort = flag.Int("port", 0, "HTTP server port")
httpPort = flag.Int("port", 0, "HTTP server port (default: chain-specific)")
pollInterval = flag.Duration("poll", 30*time.Second, "Poll interval for stats updates")
showVersion = flag.Bool("version", false, "Show version and exit")
listChains = flag.Bool("list", false, "List available chains and exit")
)
flag.Parse()
@@ -38,9 +46,27 @@ func main() {
os.Exit(0)
}
if *chainType == "" || *rpcEndpoint == "" || *databaseURL == "" || *httpPort == 0 {
if *listChains {
printChainList()
os.Exit(0)
}
if *chainType == "" {
flag.Usage()
log.Fatal("Missing required flags: -chain, -rpc, -db, -port")
fmt.Println("\nAvailable chains:")
printChainList()
os.Exit(1)
}
// Set defaults based on chain type
if *rpcEndpoint == "" {
*rpcEndpoint = defaultRPC(*chainType)
}
if *databaseURL == "" {
*databaseURL = defaultDB(*chainType)
}
if *httpPort == 0 {
*httpPort = defaultPort(*chainType)
}
ctx, cancel := context.WithCancel(context.Background())
@@ -57,87 +83,236 @@ func main() {
switch *chainType {
// DAG chains
case "xchain", "achain", "bchain", "qchain", "tchain", "zchain":
runDAGIndexer(ctx, *chainType, *rpcEndpoint, *databaseURL, *httpPort, *pollInterval)
case "xchain":
runXChain(ctx, *rpcEndpoint, *databaseURL, *httpPort, *pollInterval)
case "achain":
runAChain(ctx, *rpcEndpoint, *databaseURL, *httpPort, *pollInterval)
case "bchain":
runBChain(ctx, *rpcEndpoint, *databaseURL, *httpPort, *pollInterval)
case "qchain":
runQChain(ctx, *rpcEndpoint, *databaseURL, *httpPort, *pollInterval)
case "tchain":
runTChain(ctx, *rpcEndpoint, *databaseURL, *httpPort, *pollInterval)
case "zchain":
runZChain(ctx, *rpcEndpoint, *databaseURL, *httpPort, *pollInterval)
// Linear chains
case "pchain":
runChainIndexer(ctx, *chainType, *rpcEndpoint, *databaseURL, *httpPort, *pollInterval)
runPChain(ctx, *rpcEndpoint, *databaseURL, *httpPort, *pollInterval)
default:
log.Fatalf("Unknown chain type: %s", *chainType)
}
}
func runDAGIndexer(ctx context.Context, chainType, rpcEndpoint, databaseURL string, httpPort int, pollInterval time.Duration) {
ct := dag.ChainType(chainType)
func runXChain(ctx context.Context, rpcEndpoint, databaseURL string, httpPort int, pollInterval time.Duration) {
adapter := xchain.New(rpcEndpoint)
cfg := dag.Config{
ChainType: ct,
ChainName: chainName(chainType),
ChainType: dag.ChainX,
ChainName: "X-Chain (Exchange)",
RPCEndpoint: rpcEndpoint,
RPCMethod: rpcMethod(chainType),
RPCMethod: "xvm",
DatabaseURL: databaseURL,
HTTPPort: httpPort,
PollInterval: pollInterval,
}
// Use base adapter (chain-specific adapters can be added later)
idx, err := dag.New(cfg, nil)
idx, err := dag.New(cfg, adapter)
if err != nil {
log.Fatalf("[%s] Failed to create indexer: %v", chainType, err)
log.Fatalf("[xchain] Failed to create indexer: %v", err)
}
log.Printf("[%s] Starting DAG indexer on port %d", chainType, httpPort)
log.Printf("[xchain] Starting DAG indexer on port %d", httpPort)
if err := idx.Run(ctx); err != nil && ctx.Err() == nil {
log.Fatalf("[%s] Indexer error: %v", chainType, err)
log.Fatalf("[xchain] Indexer error: %v", err)
}
log.Printf("[%s] Indexer stopped", chainType)
log.Println("[xchain] Indexer stopped")
}
func runChainIndexer(ctx context.Context, chainType, rpcEndpoint, databaseURL string, httpPort int, pollInterval time.Duration) {
ct := chain.ChainType(chainType)
func runAChain(ctx context.Context, rpcEndpoint, databaseURL string, httpPort int, pollInterval time.Duration) {
adapter := achain.New(rpcEndpoint)
cfg := dag.Config{
ChainType: dag.ChainA,
ChainName: "A-Chain (AI)",
RPCEndpoint: rpcEndpoint,
RPCMethod: "avm",
DatabaseURL: databaseURL,
HTTPPort: httpPort,
PollInterval: pollInterval,
}
idx, err := dag.New(cfg, adapter)
if err != nil {
log.Fatalf("[achain] Failed to create indexer: %v", err)
}
log.Printf("[achain] Starting DAG indexer on port %d", httpPort)
if err := idx.Run(ctx); err != nil && ctx.Err() == nil {
log.Fatalf("[achain] Indexer error: %v", err)
}
log.Println("[achain] Indexer stopped")
}
func runBChain(ctx context.Context, rpcEndpoint, databaseURL string, httpPort int, pollInterval time.Duration) {
adapter := bchain.New(rpcEndpoint)
cfg := dag.Config{
ChainType: dag.ChainB,
ChainName: "B-Chain (Bridge)",
RPCEndpoint: rpcEndpoint,
RPCMethod: "bvm",
DatabaseURL: databaseURL,
HTTPPort: httpPort,
PollInterval: pollInterval,
}
idx, err := dag.New(cfg, adapter)
if err != nil {
log.Fatalf("[bchain] Failed to create indexer: %v", err)
}
log.Printf("[bchain] Starting DAG indexer on port %d", httpPort)
if err := idx.Run(ctx); err != nil && ctx.Err() == nil {
log.Fatalf("[bchain] Indexer error: %v", err)
}
log.Println("[bchain] Indexer stopped")
}
func runQChain(ctx context.Context, rpcEndpoint, databaseURL string, httpPort int, pollInterval time.Duration) {
adapter := qchain.New(rpcEndpoint)
cfg := dag.Config{
ChainType: dag.ChainQ,
ChainName: "Q-Chain (Quantum)",
RPCEndpoint: rpcEndpoint,
RPCMethod: "qvm",
DatabaseURL: databaseURL,
HTTPPort: httpPort,
PollInterval: pollInterval,
}
idx, err := dag.New(cfg, adapter)
if err != nil {
log.Fatalf("[qchain] Failed to create indexer: %v", err)
}
log.Printf("[qchain] Starting DAG indexer on port %d", httpPort)
if err := idx.Run(ctx); err != nil && ctx.Err() == nil {
log.Fatalf("[qchain] Indexer error: %v", err)
}
log.Println("[qchain] Indexer stopped")
}
func runTChain(ctx context.Context, rpcEndpoint, databaseURL string, httpPort int, pollInterval time.Duration) {
adapter := tchain.New(rpcEndpoint)
cfg := dag.Config{
ChainType: dag.ChainT,
ChainName: "T-Chain (Teleport)",
RPCEndpoint: rpcEndpoint,
RPCMethod: "tvm",
DatabaseURL: databaseURL,
HTTPPort: httpPort,
PollInterval: pollInterval,
}
idx, err := dag.New(cfg, adapter)
if err != nil {
log.Fatalf("[tchain] Failed to create indexer: %v", err)
}
log.Printf("[tchain] Starting DAG indexer on port %d", httpPort)
if err := idx.Run(ctx); err != nil && ctx.Err() == nil {
log.Fatalf("[tchain] Indexer error: %v", err)
}
log.Println("[tchain] Indexer stopped")
}
func runZChain(ctx context.Context, rpcEndpoint, databaseURL string, httpPort int, pollInterval time.Duration) {
adapter := zchain.New(rpcEndpoint)
cfg := dag.Config{
ChainType: dag.ChainZ,
ChainName: "Z-Chain (Privacy)",
RPCEndpoint: rpcEndpoint,
RPCMethod: "zvm",
DatabaseURL: databaseURL,
HTTPPort: httpPort,
PollInterval: pollInterval,
}
idx, err := dag.New(cfg, adapter)
if err != nil {
log.Fatalf("[zchain] Failed to create indexer: %v", err)
}
log.Printf("[zchain] Starting DAG indexer on port %d", httpPort)
if err := idx.Run(ctx); err != nil && ctx.Err() == nil {
log.Fatalf("[zchain] Indexer error: %v", err)
}
log.Println("[zchain] Indexer stopped")
}
func runPChain(ctx context.Context, rpcEndpoint, databaseURL string, httpPort int, pollInterval time.Duration) {
adapter := pchain.New(rpcEndpoint)
cfg := chain.Config{
ChainType: ct,
ChainName: chainName(chainType),
ChainType: chain.ChainP,
ChainName: "P-Chain (Platform)",
RPCEndpoint: rpcEndpoint,
RPCMethod: rpcMethod(chainType),
RPCMethod: "pvm",
DatabaseURL: databaseURL,
HTTPPort: httpPort,
PollInterval: pollInterval,
}
idx, err := chain.New(cfg, nil)
idx, err := chain.New(cfg, adapter)
if err != nil {
log.Fatalf("[%s] Failed to create indexer: %v", chainType, err)
log.Fatalf("[pchain] Failed to create indexer: %v", err)
}
log.Printf("[%s] Starting linear chain indexer on port %d", chainType, httpPort)
log.Printf("[pchain] Starting linear chain indexer on port %d", httpPort)
if err := idx.Run(ctx); err != nil && ctx.Err() == nil {
log.Fatalf("[%s] Indexer error: %v", chainType, err)
log.Fatalf("[pchain] Indexer error: %v", err)
}
log.Printf("[%s] Indexer stopped", chainType)
log.Println("[pchain] Indexer stopped")
}
func chainName(ct string) string {
names := map[string]string{
"xchain": "X-Chain (Exchange)",
"achain": "A-Chain (AI)",
"bchain": "B-Chain (Bridge)",
"qchain": "Q-Chain (Quantum)",
"tchain": "T-Chain (Teleport)",
"zchain": "Z-Chain (Privacy)",
"pchain": "P-Chain (Platform)",
}
return names[ct]
func printChainList() {
fmt.Println(" DAG-based chains (fast consensus):")
fmt.Println(" xchain - X-Chain (Exchange) - Port 4200 - Asset exchange, UTXOs")
fmt.Println(" achain - A-Chain (AI) - Port 4500 - AI compute, attestations")
fmt.Println(" bchain - B-Chain (Bridge) - Port 4600 - Cross-chain bridge")
fmt.Println(" qchain - Q-Chain (Quantum) - Port 4300 - Quantum finality proofs")
fmt.Println(" tchain - T-Chain (Teleport) - Port 4700 - MPC threshold signatures")
fmt.Println(" zchain - Z-Chain (Privacy) - Port 4400 - ZK transactions")
fmt.Println()
fmt.Println(" Linear chains (strict ordering):")
fmt.Println(" pchain - P-Chain (Platform) - Port 4100 - Validators, staking")
fmt.Println()
fmt.Println(" EVM chain (use Blockscout):")
fmt.Println(" cchain - C-Chain (Contract) - Port 4000 - Smart contracts")
}
func rpcMethod(ct string) string {
methods := map[string]string{
"xchain": "xvm",
"achain": "avm",
"bchain": "bvm",
"qchain": "qvm",
"tchain": "tvm",
"zchain": "zvm",
"pchain": "pvm",
func defaultPort(ct string) int {
ports := map[string]int{
"pchain": 4100,
"xchain": 4200,
"qchain": 4300,
"zchain": 4400,
"achain": 4500,
"bchain": 4600,
"tchain": 4700,
}
return methods[ct]
return ports[ct]
}
func defaultRPC(ct string) string {
endpoints := map[string]string{
"pchain": "http://localhost:9630/ext/bc/P",
"xchain": "http://localhost:9630/ext/bc/X",
"qchain": "http://localhost:9630/ext/bc/Q",
"zchain": "http://localhost:9630/ext/bc/Z",
"achain": "http://localhost:9630/ext/bc/A",
"bchain": "http://localhost:9630/ext/bc/B",
"tchain": "http://localhost:9630/ext/bc/T",
}
return endpoints[ct]
}
func defaultDB(ct string) string {
return fmt.Sprintf("postgres://blockscout:blockscout@localhost:5432/explorer_%s", ct)
}
+167
View File
@@ -0,0 +1,167 @@
# LUX Indexer Docker Compose
# Full stack for indexing all LUX chains (P, X, A, B, Q, T, Z)
# C-Chain uses Blockscout separately
version: '3.8'
x-indexer-common: &indexer-common
build:
context: .
dockerfile: Dockerfile
restart: unless-stopped
networks:
- hanzo-network
extra_hosts:
- "host.docker.internal:host-gateway"
depends_on:
postgres:
condition: service_healthy
services:
# PostgreSQL database for all indexers
postgres:
image: postgres:15-alpine
container_name: lux-indexer-postgres
restart: unless-stopped
environment:
POSTGRES_USER: blockscout
POSTGRES_PASSWORD: blockscout
POSTGRES_DB: indexer
volumes:
- postgres_data:/var/lib/postgresql/data
- ./scripts/init-databases.sql:/docker-entrypoint-initdb.d/init.sql
ports:
- "5433:5432"
networks:
- hanzo-network
healthcheck:
test: ["CMD-SHELL", "pg_isready -U blockscout"]
interval: 10s
timeout: 5s
retries: 5
# P-Chain Indexer (Platform - Validators, Staking)
pchain-indexer:
<<: *indexer-common
container_name: lux-pchain-indexer
command: ["-chain", "pchain", "-port", "4000"]
environment:
RPC_ENDPOINT: http://host.docker.internal:9630/ext/bc/P
DATABASE_URL: postgresql://blockscout:blockscout@postgres:5432/explorer_pchain?sslmode=disable
HTTP_PORT: 4000
ports:
- "4100:4000"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:4000/health"]
interval: 30s
timeout: 10s
retries: 3
# X-Chain Indexer (Exchange - Assets, UTXOs)
xchain-indexer:
<<: *indexer-common
container_name: lux-xchain-indexer
command: ["-chain", "xchain", "-port", "4000"]
environment:
RPC_ENDPOINT: http://host.docker.internal:9630/ext/bc/X
DATABASE_URL: postgresql://blockscout:blockscout@postgres:5432/explorer_xchain?sslmode=disable
HTTP_PORT: 4000
ports:
- "4200:4000"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:4000/health"]
interval: 30s
timeout: 10s
retries: 3
# A-Chain Indexer (AI - Attestations, Models, Inference)
achain-indexer:
<<: *indexer-common
container_name: lux-achain-indexer
command: ["-chain", "achain", "-port", "4000"]
environment:
RPC_ENDPOINT: http://host.docker.internal:9630/ext/bc/A
DATABASE_URL: postgresql://blockscout:blockscout@postgres:5432/explorer_achain?sslmode=disable
HTTP_PORT: 4000
ports:
- "4500:4000"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:4000/health"]
interval: 30s
timeout: 10s
retries: 3
# B-Chain Indexer (Bridge - Cross-chain transfers)
bchain-indexer:
<<: *indexer-common
container_name: lux-bchain-indexer
command: ["-chain", "bchain", "-port", "4000"]
environment:
RPC_ENDPOINT: http://host.docker.internal:9630/ext/bc/B
DATABASE_URL: postgresql://blockscout:blockscout@postgres:5432/explorer_bchain?sslmode=disable
HTTP_PORT: 4000
ports:
- "4600:4000"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:4000/health"]
interval: 30s
timeout: 10s
retries: 3
# Q-Chain Indexer (Quantum - Finality proofs)
qchain-indexer:
<<: *indexer-common
container_name: lux-qchain-indexer
command: ["-chain", "qchain", "-port", "4000"]
environment:
RPC_ENDPOINT: http://host.docker.internal:9630/ext/bc/Q
DATABASE_URL: postgresql://blockscout:blockscout@postgres:5432/explorer_qchain?sslmode=disable
HTTP_PORT: 4000
ports:
- "4300:4000"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:4000/health"]
interval: 30s
timeout: 10s
retries: 3
# T-Chain Indexer (Teleport - MPC signatures)
tchain-indexer:
<<: *indexer-common
container_name: lux-tchain-indexer
command: ["-chain", "tchain", "-port", "4000"]
environment:
RPC_ENDPOINT: http://host.docker.internal:9630/ext/bc/T
DATABASE_URL: postgresql://blockscout:blockscout@postgres:5432/explorer_tchain?sslmode=disable
HTTP_PORT: 4000
ports:
- "4700:4000"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:4000/health"]
interval: 30s
timeout: 10s
retries: 3
# Z-Chain Indexer (Privacy - ZK transactions)
zchain-indexer:
<<: *indexer-common
container_name: lux-zchain-indexer
command: ["-chain", "zchain", "-port", "4000"]
environment:
RPC_ENDPOINT: http://host.docker.internal:9630/ext/bc/Z
DATABASE_URL: postgresql://blockscout:blockscout@postgres:5432/explorer_zchain?sslmode=disable
HTTP_PORT: 4000
ports:
- "4400:4000"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:4000/health"]
interval: 30s
timeout: 10s
retries: 3
networks:
hanzo-network:
external: true
volumes:
postgres_data:
+558
View File
@@ -0,0 +1,558 @@
// Copyright (c) 2025 Lux Partners Limited
// SPDX-License-Identifier: MIT
// Package pchain provides the P-Chain (Platform) adapter for the LINEAR chain indexer.
// Handles validators, delegators, subnets, and staking operations.
package pchain
import (
"bytes"
"context"
"database/sql"
"encoding/json"
"fmt"
"io"
"net/http"
"time"
"github.com/luxfi/indexer/chain"
)
const (
DefaultRPCEndpoint = "http://localhost:9630/ext/bc/P"
DefaultHTTPPort = 4100
DefaultDatabaseURL = "postgres://postgres:postgres@localhost:5432/explorer_pchain?sslmode=disable"
)
// Adapter implements chain.Adapter for P-Chain
type Adapter struct {
rpcEndpoint string
httpClient *http.Client
}
// New creates a new P-Chain adapter
func New(rpcEndpoint string) *Adapter {
if rpcEndpoint == "" {
rpcEndpoint = DefaultRPCEndpoint
}
return &Adapter{
rpcEndpoint: rpcEndpoint,
httpClient: &http.Client{Timeout: 30 * time.Second},
}
}
// rpcRequest makes a JSON-RPC call to the P-Chain
func (a *Adapter) rpcRequest(ctx context.Context, method string, params interface{}) (json.RawMessage, error) {
reqBody := map[string]interface{}{
"jsonrpc": "2.0",
"id": 1,
"method": method,
"params": params,
}
body, err := json.Marshal(reqBody)
if err != nil {
return nil, fmt.Errorf("marshal request: %w", err)
}
req, err := http.NewRequestWithContext(ctx, "POST", a.rpcEndpoint, bytes.NewReader(body))
if err != nil {
return nil, fmt.Errorf("create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
resp, err := a.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("rpc call: %w", err)
}
defer resp.Body.Close()
respBody, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("read response: %w", err)
}
var rpcResp struct {
Result json.RawMessage `json:"result"`
Error *struct {
Code int `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
if err := json.Unmarshal(respBody, &rpcResp); err != nil {
return nil, fmt.Errorf("unmarshal response: %w", err)
}
if rpcResp.Error != nil {
return nil, fmt.Errorf("rpc error %d: %s", rpcResp.Error.Code, rpcResp.Error.Message)
}
return rpcResp.Result, nil
}
// PChainBlock represents a P-Chain block from RPC
type PChainBlock struct {
ID string `json:"id"`
ParentID string `json:"parentID"`
Height uint64 `json:"height"`
Timestamp int64 `json:"timestamp"` // Unix timestamp
Txs []PChainTx `json:"txs"`
}
// PChainTx represents a P-Chain transaction
type PChainTx struct {
ID string `json:"id"`
Type string `json:"type"`
Inputs json.RawMessage `json:"inputs,omitempty"`
Outputs json.RawMessage `json:"outputs,omitempty"`
Credentials json.RawMessage `json:"credentials,omitempty"`
// Validator-specific fields
NodeID string `json:"nodeID,omitempty"`
StartTime int64 `json:"startTime,omitempty"`
EndTime int64 `json:"endTime,omitempty"`
Weight uint64 `json:"weight,omitempty"`
NetID string `json:"netID,omitempty"` // For subnet operations
}
// ParseBlock implements chain.Adapter
func (a *Adapter) ParseBlock(data json.RawMessage) (*chain.Block, error) {
var pb PChainBlock
if err := json.Unmarshal(data, &pb); err != nil {
return nil, fmt.Errorf("parse pchain block: %w", err)
}
txIDs := make([]string, len(pb.Txs))
for i, tx := range pb.Txs {
txIDs[i] = tx.ID
}
metadata := map[string]interface{}{
"txTypes": extractTxTypes(pb.Txs),
}
return &chain.Block{
ID: pb.ID,
ParentID: pb.ParentID,
Height: pb.Height,
Timestamp: time.Unix(pb.Timestamp, 0),
Status: chain.StatusAccepted,
TxCount: len(pb.Txs),
TxIDs: txIDs,
Data: data,
Metadata: metadata,
}, nil
}
func extractTxTypes(txs []PChainTx) map[string]int {
types := make(map[string]int)
for _, tx := range txs {
types[tx.Type]++
}
return types
}
// GetRecentBlocks implements chain.Adapter
func (a *Adapter) GetRecentBlocks(ctx context.Context, limit int) ([]json.RawMessage, error) {
// Get current height first
heightResult, err := a.rpcRequest(ctx, "pvm.getHeight", map[string]interface{}{})
if err != nil {
return nil, fmt.Errorf("get height: %w", err)
}
var heightResp struct {
Height string `json:"height"`
}
if err := json.Unmarshal(heightResult, &heightResp); err != nil {
return nil, fmt.Errorf("parse height: %w", err)
}
var currentHeight uint64
fmt.Sscanf(heightResp.Height, "%d", &currentHeight)
blocks := make([]json.RawMessage, 0, limit)
for i := 0; i < limit && currentHeight >= uint64(i); i++ {
height := currentHeight - uint64(i)
blockData, err := a.GetBlockByHeight(ctx, height)
if err != nil {
continue // Skip failed blocks
}
blocks = append(blocks, blockData)
}
return blocks, nil
}
// GetBlockByID implements chain.Adapter
func (a *Adapter) GetBlockByID(ctx context.Context, id string) (json.RawMessage, error) {
result, err := a.rpcRequest(ctx, "pvm.getBlock", map[string]interface{}{
"blockID": id,
"encoding": "json",
})
if err != nil {
return nil, fmt.Errorf("get block by id: %w", err)
}
var blockResp struct {
Block json.RawMessage `json:"block"`
}
if err := json.Unmarshal(result, &blockResp); err != nil {
return nil, fmt.Errorf("parse block response: %w", err)
}
return blockResp.Block, nil
}
// GetBlockByHeight implements chain.Adapter
func (a *Adapter) GetBlockByHeight(ctx context.Context, height uint64) (json.RawMessage, error) {
result, err := a.rpcRequest(ctx, "pvm.getBlockByHeight", map[string]interface{}{
"height": fmt.Sprintf("%d", height),
"encoding": "json",
})
if err != nil {
return nil, fmt.Errorf("get block at height %d: %w", height, err)
}
var blockResp struct {
Block json.RawMessage `json:"block"`
}
if err := json.Unmarshal(result, &blockResp); err != nil {
return nil, fmt.Errorf("parse block response: %w", err)
}
return blockResp.Block, nil
}
// InitSchema implements chain.Adapter - creates P-Chain specific tables
func (a *Adapter) InitSchema(db *sql.DB) error {
schema := `
-- Validators table
CREATE TABLE IF NOT EXISTS pchain_validators (
node_id TEXT PRIMARY KEY,
start_time TIMESTAMPTZ NOT NULL,
end_time TIMESTAMPTZ NOT NULL,
stake_amount BIGINT NOT NULL,
potential_reward BIGINT DEFAULT 0,
delegation_fee NUMERIC(5,2) DEFAULT 0,
uptime NUMERIC(5,2) DEFAULT 0,
connected BOOLEAN DEFAULT false,
net_id TEXT DEFAULT 'primary',
tx_id TEXT,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_pchain_validators_net ON pchain_validators(net_id);
CREATE INDEX IF NOT EXISTS idx_pchain_validators_end_time ON pchain_validators(end_time);
CREATE INDEX IF NOT EXISTS idx_pchain_validators_stake ON pchain_validators(stake_amount DESC);
-- Delegators table
CREATE TABLE IF NOT EXISTS pchain_delegators (
tx_id TEXT PRIMARY KEY,
node_id TEXT NOT NULL REFERENCES pchain_validators(node_id),
start_time TIMESTAMPTZ NOT NULL,
end_time TIMESTAMPTZ NOT NULL,
stake_amount BIGINT NOT NULL,
potential_reward BIGINT DEFAULT 0,
reward_owner TEXT,
net_id TEXT DEFAULT 'primary',
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_pchain_delegators_node ON pchain_delegators(node_id);
CREATE INDEX IF NOT EXISTS idx_pchain_delegators_end ON pchain_delegators(end_time);
-- Subnets (Networks) table
CREATE TABLE IF NOT EXISTS pchain_nets (
net_id TEXT PRIMARY KEY,
owner_addresses JSONB DEFAULT '[]',
threshold INT DEFAULT 1,
control_keys JSONB DEFAULT '[]',
tx_id TEXT,
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- Blockchains table
CREATE TABLE IF NOT EXISTS pchain_chains (
chain_id TEXT PRIMARY KEY,
net_id TEXT REFERENCES pchain_nets(net_id),
name TEXT NOT NULL,
vm_id TEXT NOT NULL,
genesis_data TEXT,
tx_id TEXT,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_pchain_chains_net ON pchain_chains(net_id);
-- Staking rewards table
CREATE TABLE IF NOT EXISTS pchain_rewards (
id SERIAL PRIMARY KEY,
tx_id TEXT NOT NULL,
node_id TEXT,
delegator_tx_id TEXT,
amount BIGINT NOT NULL,
reward_type TEXT NOT NULL,
claimed_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_pchain_rewards_node ON pchain_rewards(node_id);
CREATE INDEX IF NOT EXISTS idx_pchain_rewards_time ON pchain_rewards(claimed_at DESC);
-- P-Chain transactions table (extended)
CREATE TABLE IF NOT EXISTS pchain_txs (
tx_id TEXT PRIMARY KEY,
block_id TEXT,
tx_type TEXT NOT NULL,
inputs JSONB DEFAULT '[]',
outputs JSONB DEFAULT '[]',
memo TEXT,
fee BIGINT DEFAULT 0,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_pchain_txs_type ON pchain_txs(tx_type);
CREATE INDEX IF NOT EXISTS idx_pchain_txs_block ON pchain_txs(block_id);
-- Extended stats for P-Chain
CREATE TABLE IF NOT EXISTS pchain_extended_stats (
id INT PRIMARY KEY DEFAULT 1,
total_validators INT DEFAULT 0,
active_validators INT DEFAULT 0,
total_delegators INT DEFAULT 0,
total_stake BIGINT DEFAULT 0,
total_delegated BIGINT DEFAULT 0,
total_nets INT DEFAULT 0,
total_chains INT DEFAULT 0,
updated_at TIMESTAMPTZ DEFAULT NOW()
);
INSERT INTO pchain_extended_stats (id) VALUES (1) ON CONFLICT DO NOTHING;
`
if _, err := db.Exec(schema); err != nil {
return fmt.Errorf("pchain schema: %w", err)
}
return nil
}
// GetStats implements chain.Adapter - returns P-Chain specific statistics
func (a *Adapter) GetStats(ctx context.Context, db *sql.DB) (map[string]interface{}, error) {
stats := make(map[string]interface{})
// Get validator stats
var totalValidators, activeValidators int
var totalStake int64
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM pchain_validators").Scan(&totalValidators)
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM pchain_validators WHERE end_time > NOW()").Scan(&activeValidators)
db.QueryRowContext(ctx, "SELECT COALESCE(SUM(stake_amount), 0) FROM pchain_validators WHERE end_time > NOW()").Scan(&totalStake)
stats["validators"] = map[string]interface{}{
"total": totalValidators,
"active": activeValidators,
"total_stake": totalStake,
}
// Get delegator stats
var totalDelegators int
var totalDelegated int64
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM pchain_delegators WHERE end_time > NOW()").Scan(&totalDelegators)
db.QueryRowContext(ctx, "SELECT COALESCE(SUM(stake_amount), 0) FROM pchain_delegators WHERE end_time > NOW()").Scan(&totalDelegated)
stats["delegators"] = map[string]interface{}{
"active": totalDelegators,
"total_delegated": totalDelegated,
}
// Get subnet/network stats
var totalNets, totalChains int
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM pchain_nets").Scan(&totalNets)
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM pchain_chains").Scan(&totalChains)
stats["networks"] = map[string]interface{}{
"total_nets": totalNets,
"total_chains": totalChains,
}
// Update extended stats
db.ExecContext(ctx, `
UPDATE pchain_extended_stats SET
total_validators = $1,
active_validators = $2,
total_delegators = $3,
total_stake = $4,
total_delegated = $5,
total_nets = $6,
total_chains = $7,
updated_at = NOW()
WHERE id = 1
`, totalValidators, activeValidators, totalDelegators, totalStake, totalDelegated, totalNets, totalChains)
return stats, nil
}
// GetCurrentValidators fetches current validators from RPC
func (a *Adapter) GetCurrentValidators(ctx context.Context, netID string) ([]Validator, error) {
params := map[string]interface{}{}
if netID != "" && netID != "primary" {
params["netID"] = netID
}
result, err := a.rpcRequest(ctx, "pvm.getCurrentValidators", params)
if err != nil {
return nil, fmt.Errorf("get current validators: %w", err)
}
var resp struct {
Validators []Validator `json:"validators"`
}
if err := json.Unmarshal(result, &resp); err != nil {
return nil, fmt.Errorf("parse validators: %w", err)
}
return resp.Validators, nil
}
// GetPendingValidators fetches pending validators from RPC
func (a *Adapter) GetPendingValidators(ctx context.Context, netID string) ([]Validator, error) {
params := map[string]interface{}{}
if netID != "" && netID != "primary" {
params["netID"] = netID
}
result, err := a.rpcRequest(ctx, "pvm.getPendingValidators", params)
if err != nil {
return nil, fmt.Errorf("get pending validators: %w", err)
}
var resp struct {
Validators []Validator `json:"validators"`
}
if err := json.Unmarshal(result, &resp); err != nil {
return nil, fmt.Errorf("parse pending validators: %w", err)
}
return resp.Validators, nil
}
// Validator represents a P-Chain validator
type Validator struct {
TxID string `json:"txID"`
NodeID string `json:"nodeID"`
StartTime string `json:"startTime"`
EndTime string `json:"endTime"`
StakeAmount string `json:"stakeAmount"`
PotentialReward string `json:"potentialReward"`
DelegationFee string `json:"delegationFee"`
Uptime string `json:"uptime"`
Connected bool `json:"connected"`
Delegators []Delegator `json:"delegators,omitempty"`
}
// Delegator represents a P-Chain delegator
type Delegator struct {
TxID string `json:"txID"`
NodeID string `json:"nodeID"`
StartTime string `json:"startTime"`
EndTime string `json:"endTime"`
StakeAmount string `json:"stakeAmount"`
PotentialReward string `json:"potentialReward"`
RewardOwner string `json:"rewardOwner,omitempty"`
}
// GetNets fetches all subnets/networks from RPC
func (a *Adapter) GetNets(ctx context.Context) ([]Net, error) {
result, err := a.rpcRequest(ctx, "pvm.getNets", map[string]interface{}{})
if err != nil {
return nil, fmt.Errorf("get nets: %w", err)
}
var resp struct {
Nets []Net `json:"nets"`
}
if err := json.Unmarshal(result, &resp); err != nil {
return nil, fmt.Errorf("parse nets: %w", err)
}
return resp.Nets, nil
}
// Net represents a P-Chain subnet/network
type Net struct {
ID string `json:"id"`
ControlKeys []string `json:"controlKeys"`
Threshold string `json:"threshold"`
}
// GetBlockchains fetches all blockchains from RPC
func (a *Adapter) GetBlockchains(ctx context.Context) ([]Blockchain, error) {
result, err := a.rpcRequest(ctx, "pvm.getBlockchains", map[string]interface{}{})
if err != nil {
return nil, fmt.Errorf("get blockchains: %w", err)
}
var resp struct {
Blockchains []Blockchain `json:"blockchains"`
}
if err := json.Unmarshal(result, &resp); err != nil {
return nil, fmt.Errorf("parse blockchains: %w", err)
}
return resp.Blockchains, nil
}
// Blockchain represents a P-Chain blockchain
type Blockchain struct {
ID string `json:"id"`
Name string `json:"name"`
NetID string `json:"netID"`
VMID string `json:"vmID"`
}
// SyncValidators syncs validator data from RPC to database
func (a *Adapter) SyncValidators(ctx context.Context, db *sql.DB) error {
validators, err := a.GetCurrentValidators(ctx, "")
if err != nil {
return err
}
for _, v := range validators {
_, err := db.ExecContext(ctx, `
INSERT INTO pchain_validators (node_id, start_time, end_time, stake_amount, potential_reward, delegation_fee, uptime, connected, tx_id, updated_at)
VALUES ($1, to_timestamp($2::bigint), to_timestamp($3::bigint), $4, $5, $6, $7, $8, $9, NOW())
ON CONFLICT (node_id) DO UPDATE SET
end_time = EXCLUDED.end_time,
stake_amount = EXCLUDED.stake_amount,
potential_reward = EXCLUDED.potential_reward,
uptime = EXCLUDED.uptime,
connected = EXCLUDED.connected,
updated_at = NOW()
`, v.NodeID, v.StartTime, v.EndTime, v.StakeAmount, v.PotentialReward, v.DelegationFee, v.Uptime, v.Connected, v.TxID)
if err != nil {
return fmt.Errorf("upsert validator %s: %w", v.NodeID, err)
}
// Sync delegators for this validator
for _, d := range v.Delegators {
_, err := db.ExecContext(ctx, `
INSERT INTO pchain_delegators (tx_id, node_id, start_time, end_time, stake_amount, potential_reward, reward_owner)
VALUES ($1, $2, to_timestamp($3::bigint), to_timestamp($4::bigint), $5, $6, $7)
ON CONFLICT (tx_id) DO UPDATE SET
end_time = EXCLUDED.end_time,
potential_reward = EXCLUDED.potential_reward
`, d.TxID, d.NodeID, d.StartTime, d.EndTime, d.StakeAmount, d.PotentialReward, d.RewardOwner)
if err != nil {
return fmt.Errorf("upsert delegator %s: %w", d.TxID, err)
}
}
}
return nil
}
// NewConfig creates a default P-Chain indexer configuration
func NewConfig() chain.Config {
return chain.Config{
ChainType: chain.ChainP,
ChainName: "P-Chain (Platform)",
RPCEndpoint: DefaultRPCEndpoint,
RPCMethod: "pvm",
DatabaseURL: DefaultDatabaseURL,
HTTPPort: DefaultHTTPPort,
PollInterval: 5 * time.Second,
}
}
// Ensure Adapter implements chain.Adapter
var _ chain.Adapter = (*Adapter)(nil)
+556
View File
@@ -0,0 +1,556 @@
// Copyright (c) 2025 Lux Partners Limited
// SPDX-License-Identifier: MIT
// Package qchain provides the Q-Chain (Quantum) adapter for the DAG indexer.
// Q-Chain handles quantum-resistant finality proofs using lattice-based cryptography.
package qchain
import (
"bytes"
"context"
"database/sql"
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/luxfi/indexer/dag"
)
const (
// DefaultRPCEndpoint is the default Q-Chain RPC endpoint
DefaultRPCEndpoint = "http://localhost:9630/ext/bc/Q/rpc"
// DefaultHTTPPort is the default API port for Q-Chain indexer
DefaultHTTPPort = 4300
// DefaultDatabaseURL is the default PostgreSQL connection
DefaultDatabaseURL = "postgres://blockscout:blockscout@localhost:5432/explorer_qchain?sslmode=disable"
)
// FinalityProof represents a quantum-resistant finality proof
type FinalityProof struct {
ID string `json:"id"`
VertexID string `json:"vertexId"`
ProofType ProofType `json:"proofType"`
LatticeParams Lattice `json:"latticeParams"`
Signature []byte `json:"signature"`
PublicKey []byte `json:"publicKey"`
Timestamp time.Time `json:"timestamp"`
Verified bool `json:"verified"`
}
// ProofType identifies the quantum-resistant algorithm used
type ProofType string
const (
ProofDilithium ProofType = "dilithium" // NIST PQC standard
ProofKyber ProofType = "kyber" // Key encapsulation
ProofFalcon ProofType = "falcon" // Fast lattice-based
ProofSphincsh ProofType = "sphincs+" // Hash-based
)
// Lattice contains lattice-based cryptography parameters
type Lattice struct {
Dimension int `json:"dimension"` // Lattice dimension (n)
Modulus int64 `json:"modulus"` // Modulus (q)
ErrorBound int `json:"errorBound"` // Error distribution bound
SecurityLvl int `json:"securityLvl"` // NIST security level (1-5)
Algorithm string `json:"algorithm"` // Specific algorithm variant
}
// QuantumVertex extends DAG vertex with quantum-specific data
type QuantumVertex struct {
dag.Vertex
FinalityProof *FinalityProof `json:"finalityProof,omitempty"`
ProofCount int `json:"proofCount"`
Finalized bool `json:"finalized"`
}
// QuantumStamp represents a quantum-certified timestamp for finality
type QuantumStamp struct {
ID string `json:"id"`
VertexID string `json:"vertexId"`
ChainID string `json:"chainId"` // Which chain this stamp is for
BlockHeight uint64 `json:"blockHeight"` // Block height being stamped
BlockHash []byte `json:"blockHash"` // Block hash being certified
Entropy []byte `json:"entropy"` // Quantum random entropy
KeyID string `json:"keyId"` // Corona key used for signing
Signature []byte `json:"signature"` // Quantum-resistant signature
Timestamp time.Time `json:"timestamp"`
Certified bool `json:"certified"` // Whether stamp has been verified
}
// CoronaKey represents a quantum-resistant signing key (Corona is LUX's post-quantum signature scheme)
type CoronaKey struct {
ID string `json:"id"`
PublicKey []byte `json:"publicKey"`
KeyType ProofType `json:"keyType"` // dilithium, falcon, sphincs+
Algorithm string `json:"algorithm"` // Specific algorithm variant
SecurityLvl int `json:"securityLevel"` // NIST security level 1-5
Owner string `json:"owner"` // Key owner address
ValidFrom time.Time `json:"validFrom"`
ValidUntil time.Time `json:"validUntil"`
Revoked bool `json:"revoked"`
CreatedAt time.Time `json:"createdAt"`
}
// Adapter implements dag.Adapter for Q-Chain
type Adapter struct {
rpcEndpoint string
httpClient *http.Client
}
// New creates a new Q-Chain adapter
func New(rpcEndpoint string) *Adapter {
if rpcEndpoint == "" {
rpcEndpoint = DefaultRPCEndpoint
}
return &Adapter{
rpcEndpoint: rpcEndpoint,
httpClient: &http.Client{Timeout: 30 * time.Second},
}
}
// rpcRequest represents a JSON-RPC request
type rpcRequest struct {
JSONRPC string `json:"jsonrpc"`
ID int `json:"id"`
Method string `json:"method"`
Params []interface{} `json:"params,omitempty"`
}
// rpcResponse represents a JSON-RPC response
type rpcResponse struct {
JSONRPC string `json:"jsonrpc"`
ID int `json:"id"`
Result json.RawMessage `json:"result,omitempty"`
Error *rpcError `json:"error,omitempty"`
}
type rpcError struct {
Code int `json:"code"`
Message string `json:"message"`
}
// call executes an RPC request to the Q-Chain node
func (a *Adapter) call(ctx context.Context, method string, params ...interface{}) (json.RawMessage, error) {
req := rpcRequest{
JSONRPC: "2.0",
ID: 1,
Method: method,
Params: params,
}
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("marshal request: %w", err)
}
httpReq, err := http.NewRequestWithContext(ctx, "POST", a.rpcEndpoint, bytes.NewReader(body))
if err != nil {
return nil, fmt.Errorf("create request: %w", err)
}
httpReq.Header.Set("Content-Type", "application/json")
resp, err := a.httpClient.Do(httpReq)
if err != nil {
return nil, fmt.Errorf("http request: %w", err)
}
defer resp.Body.Close()
var rpcResp rpcResponse
if err := json.NewDecoder(resp.Body).Decode(&rpcResp); err != nil {
return nil, fmt.Errorf("decode response: %w", err)
}
if rpcResp.Error != nil {
return nil, fmt.Errorf("rpc error %d: %s", rpcResp.Error.Code, rpcResp.Error.Message)
}
return rpcResp.Result, nil
}
// ParseVertex parses raw RPC data into a DAG vertex
func (a *Adapter) ParseVertex(data json.RawMessage) (*dag.Vertex, error) {
var qv QuantumVertex
if err := json.Unmarshal(data, &qv); err != nil {
return nil, fmt.Errorf("unmarshal quantum vertex: %w", err)
}
// Build metadata with quantum-specific fields
meta := make(map[string]interface{})
if qv.FinalityProof != nil {
meta["proof_type"] = qv.FinalityProof.ProofType
meta["proof_verified"] = qv.FinalityProof.Verified
meta["lattice_dimension"] = qv.FinalityProof.LatticeParams.Dimension
meta["security_level"] = qv.FinalityProof.LatticeParams.SecurityLvl
}
meta["proof_count"] = qv.ProofCount
meta["finalized"] = qv.Finalized
v := &dag.Vertex{
ID: qv.ID,
Type: "quantum",
ParentIDs: qv.ParentIDs,
Height: qv.Height,
Epoch: qv.Epoch,
TxIDs: qv.TxIDs,
Timestamp: qv.Timestamp,
Status: qv.Status,
Data: data,
Metadata: meta,
}
// Update status based on finalization
if qv.Finalized {
v.Status = dag.StatusAccepted
}
return v, nil
}
// GetRecentVertices fetches recent vertices from Q-Chain
func (a *Adapter) GetRecentVertices(ctx context.Context, limit int) ([]json.RawMessage, error) {
result, err := a.call(ctx, "qvm.getRecentVertices", map[string]interface{}{
"limit": limit,
})
if err != nil {
return nil, err
}
var response struct {
Vertices []json.RawMessage `json:"vertices"`
}
if err := json.Unmarshal(result, &response); err != nil {
return nil, fmt.Errorf("unmarshal vertices: %w", err)
}
return response.Vertices, nil
}
// GetVertexByID fetches a specific vertex by ID
func (a *Adapter) GetVertexByID(ctx context.Context, id string) (json.RawMessage, error) {
result, err := a.call(ctx, "qvm.getVertex", map[string]interface{}{
"id": id,
})
if err != nil {
return nil, err
}
var response struct {
Vertex json.RawMessage `json:"vertex"`
}
if err := json.Unmarshal(result, &response); err != nil {
return nil, fmt.Errorf("unmarshal vertex: %w", err)
}
return response.Vertex, nil
}
// InitSchema creates Q-Chain specific database tables
func (a *Adapter) InitSchema(db *sql.DB) error {
schema := `
-- Finality proofs table
CREATE TABLE IF NOT EXISTS qchain_finality_proofs (
id TEXT PRIMARY KEY,
vertex_id TEXT NOT NULL REFERENCES qchain_vertices(id),
proof_type TEXT NOT NULL,
lattice_dimension INT,
lattice_modulus BIGINT,
lattice_error_bound INT,
security_level INT,
algorithm TEXT,
signature BYTEA,
public_key BYTEA,
verified BOOLEAN DEFAULT FALSE,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_qchain_proofs_vertex ON qchain_finality_proofs(vertex_id);
CREATE INDEX IF NOT EXISTS idx_qchain_proofs_type ON qchain_finality_proofs(proof_type);
CREATE INDEX IF NOT EXISTS idx_qchain_proofs_verified ON qchain_finality_proofs(verified);
-- Lattice parameters table (for reusable lattice configs)
CREATE TABLE IF NOT EXISTS qchain_lattice_params (
id SERIAL PRIMARY KEY,
name TEXT UNIQUE NOT NULL,
dimension INT NOT NULL,
modulus BIGINT NOT NULL,
error_bound INT NOT NULL,
security_level INT NOT NULL,
algorithm TEXT NOT NULL,
description TEXT,
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- Insert default lattice configurations
INSERT INTO qchain_lattice_params (name, dimension, modulus, error_bound, security_level, algorithm, description)
VALUES
('dilithium2', 256, 8380417, 2, 2, 'dilithium', 'NIST Level 2 Dilithium'),
('dilithium3', 256, 8380417, 4, 3, 'dilithium', 'NIST Level 3 Dilithium'),
('dilithium5', 256, 8380417, 8, 5, 'dilithium', 'NIST Level 5 Dilithium'),
('kyber512', 256, 3329, 2, 1, 'kyber', 'Kyber-512'),
('kyber768', 256, 3329, 2, 3, 'kyber', 'Kyber-768'),
('kyber1024', 256, 3329, 2, 5, 'kyber', 'Kyber-1024'),
('falcon512', 512, 12289, 1, 1, 'falcon', 'Falcon-512'),
('falcon1024', 1024, 12289, 1, 5, 'falcon', 'Falcon-1024')
ON CONFLICT (name) DO NOTHING;
-- Quantum stamps for cross-chain finality
CREATE TABLE IF NOT EXISTS qchain_stamps (
id TEXT PRIMARY KEY,
vertex_id TEXT,
chain_id TEXT NOT NULL,
block_height BIGINT NOT NULL,
block_hash BYTEA NOT NULL,
entropy BYTEA,
key_id TEXT,
signature BYTEA,
timestamp TIMESTAMPTZ NOT NULL,
certified BOOLEAN DEFAULT FALSE,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_qchain_stamps_chain ON qchain_stamps(chain_id);
CREATE INDEX IF NOT EXISTS idx_qchain_stamps_height ON qchain_stamps(chain_id, block_height);
CREATE INDEX IF NOT EXISTS idx_qchain_stamps_key ON qchain_stamps(key_id);
CREATE INDEX IF NOT EXISTS idx_qchain_stamps_certified ON qchain_stamps(certified);
-- Corona quantum-resistant keys
CREATE TABLE IF NOT EXISTS qchain_corona_keys (
id TEXT PRIMARY KEY,
public_key BYTEA NOT NULL,
key_type TEXT NOT NULL,
algorithm TEXT NOT NULL,
security_level INT NOT NULL,
owner TEXT NOT NULL,
valid_from TIMESTAMPTZ NOT NULL,
valid_until TIMESTAMPTZ NOT NULL,
revoked BOOLEAN DEFAULT FALSE,
vertex_id TEXT,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_qchain_keys_owner ON qchain_corona_keys(owner);
CREATE INDEX IF NOT EXISTS idx_qchain_keys_type ON qchain_corona_keys(key_type);
CREATE INDEX IF NOT EXISTS idx_qchain_keys_active ON qchain_corona_keys(revoked, valid_until);
-- Q-Chain extended stats
CREATE TABLE IF NOT EXISTS qchain_extended_stats (
id INT PRIMARY KEY DEFAULT 1,
total_proofs BIGINT DEFAULT 0,
verified_proofs BIGINT DEFAULT 0,
finalized_vertices BIGINT DEFAULT 0,
total_stamps BIGINT DEFAULT 0,
certified_stamps BIGINT DEFAULT 0,
total_keys BIGINT DEFAULT 0,
active_keys BIGINT DEFAULT 0,
dilithium_proofs BIGINT DEFAULT 0,
kyber_proofs BIGINT DEFAULT 0,
falcon_proofs BIGINT DEFAULT 0,
sphincs_proofs BIGINT DEFAULT 0,
avg_security_level FLOAT DEFAULT 0,
updated_at TIMESTAMPTZ DEFAULT NOW()
);
INSERT INTO qchain_extended_stats (id) VALUES (1) ON CONFLICT DO NOTHING;
`
if _, err := db.Exec(schema); err != nil {
return fmt.Errorf("init qchain schema: %w", err)
}
return nil
}
// GetStats returns Q-Chain specific statistics
func (a *Adapter) GetStats(ctx context.Context, db *sql.DB) (map[string]interface{}, error) {
stats := make(map[string]interface{})
// Query extended stats
var totalProofs, verifiedProofs, finalizedVertices int64
var dilithiumProofs, kyberProofs, falconProofs, sphincsProofs int64
var avgSecurityLevel float64
row := db.QueryRowContext(ctx, `
SELECT total_proofs, verified_proofs, finalized_vertices,
dilithium_proofs, kyber_proofs, falcon_proofs, sphincs_proofs,
avg_security_level
FROM qchain_extended_stats WHERE id = 1
`)
if err := row.Scan(
&totalProofs, &verifiedProofs, &finalizedVertices,
&dilithiumProofs, &kyberProofs, &falconProofs, &sphincsProofs,
&avgSecurityLevel,
); err != nil && err != sql.ErrNoRows {
return nil, fmt.Errorf("query extended stats: %w", err)
}
stats["total_proofs"] = totalProofs
stats["verified_proofs"] = verifiedProofs
stats["finalized_vertices"] = finalizedVertices
stats["proof_types"] = map[string]int64{
"dilithium": dilithiumProofs,
"kyber": kyberProofs,
"falcon": falconProofs,
"sphincs": sphincsProofs,
}
stats["avg_security_level"] = avgSecurityLevel
// Calculate verification rate
if totalProofs > 0 {
stats["verification_rate"] = float64(verifiedProofs) / float64(totalProofs)
} else {
stats["verification_rate"] = 0.0
}
return stats, nil
}
// StoreProof stores a finality proof in the database
func (a *Adapter) StoreProof(ctx context.Context, db *sql.DB, proof *FinalityProof) error {
_, err := db.ExecContext(ctx, `
INSERT INTO qchain_finality_proofs
(id, vertex_id, proof_type, lattice_dimension, lattice_modulus,
lattice_error_bound, security_level, algorithm, signature, public_key, verified)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
ON CONFLICT (id) DO UPDATE SET verified = EXCLUDED.verified
`,
proof.ID, proof.VertexID, proof.ProofType,
proof.LatticeParams.Dimension, proof.LatticeParams.Modulus,
proof.LatticeParams.ErrorBound, proof.LatticeParams.SecurityLvl,
proof.LatticeParams.Algorithm, proof.Signature, proof.PublicKey, proof.Verified,
)
return err
}
// VerifyProof performs quantum-resistant proof verification
// Returns true if the proof is valid under the specified lattice parameters
func (a *Adapter) VerifyProof(ctx context.Context, proof *FinalityProof) (bool, error) {
// Delegate verification to the Q-Chain node via RPC
result, err := a.call(ctx, "qvm.verifyProof", map[string]interface{}{
"proofId": proof.ID,
"vertexId": proof.VertexID,
"proofType": proof.ProofType,
"signature": proof.Signature,
"publicKey": proof.PublicKey,
})
if err != nil {
return false, err
}
var response struct {
Valid bool `json:"valid"`
Message string `json:"message,omitempty"`
}
if err := json.Unmarshal(result, &response); err != nil {
return false, fmt.Errorf("unmarshal verify response: %w", err)
}
return response.Valid, nil
}
// UpdateExtendedStats updates Q-Chain specific statistics
func (a *Adapter) UpdateExtendedStats(ctx context.Context, db *sql.DB) error {
_, err := db.ExecContext(ctx, `
UPDATE qchain_extended_stats SET
total_proofs = (SELECT COUNT(*) FROM qchain_finality_proofs),
verified_proofs = (SELECT COUNT(*) FROM qchain_finality_proofs WHERE verified = TRUE),
finalized_vertices = (SELECT COUNT(*) FROM qchain_vertices WHERE status = 'accepted'),
dilithium_proofs = (SELECT COUNT(*) FROM qchain_finality_proofs WHERE proof_type = 'dilithium'),
kyber_proofs = (SELECT COUNT(*) FROM qchain_finality_proofs WHERE proof_type = 'kyber'),
falcon_proofs = (SELECT COUNT(*) FROM qchain_finality_proofs WHERE proof_type = 'falcon'),
sphincs_proofs = (SELECT COUNT(*) FROM qchain_finality_proofs WHERE proof_type = 'sphincs+'),
avg_security_level = COALESCE((SELECT AVG(security_level) FROM qchain_finality_proofs), 0),
updated_at = NOW()
WHERE id = 1
`)
return err
}
// NewConfig creates a default Q-Chain indexer configuration
func NewConfig() dag.Config {
return dag.Config{
ChainType: dag.ChainQ,
ChainName: "Q-Chain (Quantum)",
RPCEndpoint: DefaultRPCEndpoint,
RPCMethod: "qvm",
DatabaseURL: DefaultDatabaseURL,
HTTPPort: DefaultHTTPPort,
PollInterval: 5 * time.Second,
}
}
// StoreStamp stores a quantum stamp for cross-chain finality
func (a *Adapter) StoreStamp(ctx context.Context, db *sql.DB, stamp *QuantumStamp) error {
_, err := db.ExecContext(ctx, `
INSERT INTO qchain_stamps
(id, vertex_id, chain_id, block_height, block_hash, entropy, key_id, signature, timestamp, certified)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
ON CONFLICT (id) DO UPDATE SET certified = EXCLUDED.certified
`, stamp.ID, stamp.VertexID, stamp.ChainID, stamp.BlockHeight, stamp.BlockHash,
stamp.Entropy, stamp.KeyID, stamp.Signature, stamp.Timestamp, stamp.Certified)
return err
}
// StoreCoronaKey stores a quantum-resistant signing key
func (a *Adapter) StoreCoronaKey(ctx context.Context, db *sql.DB, key *CoronaKey, vertexID string) error {
_, err := db.ExecContext(ctx, `
INSERT INTO qchain_corona_keys
(id, public_key, key_type, algorithm, security_level, owner, valid_from, valid_until, revoked, vertex_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
ON CONFLICT (id) DO UPDATE SET revoked = EXCLUDED.revoked
`, key.ID, key.PublicKey, key.KeyType, key.Algorithm, key.SecurityLvl,
key.Owner, key.ValidFrom, key.ValidUntil, key.Revoked, vertexID)
return err
}
// GetStampsByChain retrieves quantum stamps for a specific chain
func (a *Adapter) GetStampsByChain(ctx context.Context, db *sql.DB, chainID string, limit int) ([]QuantumStamp, error) {
rows, err := db.QueryContext(ctx, `
SELECT id, vertex_id, chain_id, block_height, block_hash, entropy, key_id, signature, timestamp, certified
FROM qchain_stamps
WHERE chain_id = $1
ORDER BY block_height DESC
LIMIT $2
`, chainID, limit)
if err != nil {
return nil, err
}
defer rows.Close()
var stamps []QuantumStamp
for rows.Next() {
var s QuantumStamp
if err := rows.Scan(&s.ID, &s.VertexID, &s.ChainID, &s.BlockHeight, &s.BlockHash,
&s.Entropy, &s.KeyID, &s.Signature, &s.Timestamp, &s.Certified); err != nil {
continue
}
stamps = append(stamps, s)
}
return stamps, nil
}
// GetActiveKeys retrieves active Corona keys for an owner
func (a *Adapter) GetActiveKeys(ctx context.Context, db *sql.DB, owner string) ([]CoronaKey, error) {
rows, err := db.QueryContext(ctx, `
SELECT id, public_key, key_type, algorithm, security_level, owner, valid_from, valid_until, revoked, created_at
FROM qchain_corona_keys
WHERE owner = $1 AND revoked = FALSE AND valid_until > NOW()
ORDER BY created_at DESC
`, owner)
if err != nil {
return nil, err
}
defer rows.Close()
var keys []CoronaKey
for rows.Next() {
var k CoronaKey
if err := rows.Scan(&k.ID, &k.PublicKey, &k.KeyType, &k.Algorithm, &k.SecurityLvl,
&k.Owner, &k.ValidFrom, &k.ValidUntil, &k.Revoked, &k.CreatedAt); err != nil {
continue
}
keys = append(keys, k)
}
return keys, nil
}
// Verify interface compliance at compile time
var _ dag.Adapter = (*Adapter)(nil)
+52
View File
@@ -0,0 +1,52 @@
-- LUX Indexer Database Initialization
-- Creates databases for all chain indexers
-- P-Chain (Platform)
CREATE DATABASE explorer_pchain;
GRANT ALL PRIVILEGES ON DATABASE explorer_pchain TO blockscout;
-- X-Chain (Exchange)
CREATE DATABASE explorer_xchain;
GRANT ALL PRIVILEGES ON DATABASE explorer_xchain TO blockscout;
-- A-Chain (AI)
CREATE DATABASE explorer_achain;
GRANT ALL PRIVILEGES ON DATABASE explorer_achain TO blockscout;
-- B-Chain (Bridge)
CREATE DATABASE explorer_bchain;
GRANT ALL PRIVILEGES ON DATABASE explorer_bchain TO blockscout;
-- Q-Chain (Quantum)
CREATE DATABASE explorer_qchain;
GRANT ALL PRIVILEGES ON DATABASE explorer_qchain TO blockscout;
-- T-Chain (Teleport)
CREATE DATABASE explorer_tchain;
GRANT ALL PRIVILEGES ON DATABASE explorer_tchain TO blockscout;
-- Z-Chain (Privacy)
CREATE DATABASE explorer_zchain;
GRANT ALL PRIVILEGES ON DATABASE explorer_zchain TO blockscout;
-- Testnet databases
CREATE DATABASE explorer_pchain_test;
GRANT ALL PRIVILEGES ON DATABASE explorer_pchain_test TO blockscout;
CREATE DATABASE explorer_xchain_test;
GRANT ALL PRIVILEGES ON DATABASE explorer_xchain_test TO blockscout;
CREATE DATABASE explorer_achain_test;
GRANT ALL PRIVILEGES ON DATABASE explorer_achain_test TO blockscout;
CREATE DATABASE explorer_bchain_test;
GRANT ALL PRIVILEGES ON DATABASE explorer_bchain_test TO blockscout;
CREATE DATABASE explorer_qchain_test;
GRANT ALL PRIVILEGES ON DATABASE explorer_qchain_test TO blockscout;
CREATE DATABASE explorer_tchain_test;
GRANT ALL PRIVILEGES ON DATABASE explorer_tchain_test TO blockscout;
CREATE DATABASE explorer_zchain_test;
GRANT ALL PRIVILEGES ON DATABASE explorer_zchain_test TO blockscout;
+635
View File
@@ -0,0 +1,635 @@
// Copyright (c) 2025 Lux Partners Limited
// SPDX-License-Identifier: MIT
// Package tchain provides T-Chain (Teleport) adapter for DAG indexing.
// T-Chain handles MPC threshold signatures for cross-chain teleportation.
package tchain
import (
"bytes"
"context"
"database/sql"
"encoding/hex"
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/luxfi/indexer/dag"
)
const (
// DefaultRPCEndpoint for T-Chain on port 4700
DefaultRPCEndpoint = "http://localhost:9630/ext/bc/T/rpc"
// DefaultHTTPPort for T-Chain indexer API
DefaultHTTPPort = 4700
// DefaultDatabaseURL for T-Chain explorer
DefaultDatabaseURL = "postgres://blockscout:blockscout@localhost:5432/explorer_tchain?sslmode=disable"
)
// SessionStatus for MPC signing sessions
type SessionStatus string
const (
SessionPending SessionStatus = "pending"
SessionActive SessionStatus = "active"
SessionCompleted SessionStatus = "completed"
SessionFailed SessionStatus = "failed"
SessionExpired SessionStatus = "expired"
)
// KeyShareStatus for distributed key shares
type KeyShareStatus string
const (
SharePending KeyShareStatus = "pending"
ShareActive KeyShareStatus = "active"
ShareRevoked KeyShareStatus = "revoked"
ShareRotated KeyShareStatus = "rotated"
)
// SigningSession represents an MPC threshold signing session
type SigningSession struct {
ID string `json:"id"`
Threshold uint32 `json:"threshold"` // t of n threshold
TotalShares uint32 `json:"totalShares"` // total participants
MessageHash string `json:"messageHash"` // hash being signed
Participants []string `json:"participants"` // node IDs participating
Signatures []string `json:"signatures"` // partial signatures collected
FinalSig string `json:"finalSig,omitempty"` // combined signature
Status SessionStatus `json:"status"`
CreatedAt time.Time `json:"createdAt"`
CompletedAt *time.Time `json:"completedAt,omitempty"`
ExpiresAt time.Time `json:"expiresAt"`
SourceChain string `json:"sourceChain,omitempty"`
DestChain string `json:"destChain,omitempty"`
}
// KeyShare represents a distributed key share for MPC
type KeyShare struct {
ID string `json:"id"`
PublicKey string `json:"publicKey"`
ShareIndex uint32 `json:"shareIndex"`
NodeID string `json:"nodeId"`
KeyGenID string `json:"keyGenId"` // key generation ceremony ID
Status KeyShareStatus `json:"status"`
CreatedAt time.Time `json:"createdAt"`
RotatedAt *time.Time `json:"rotatedAt,omitempty"`
ExpiresAt *time.Time `json:"expiresAt,omitempty"`
}
// KeyGeneration represents a distributed key generation ceremony
type KeyGeneration struct {
ID string `json:"id"`
Threshold uint32 `json:"threshold"`
TotalShares uint32 `json:"totalShares"`
PublicKey string `json:"publicKey"` // combined public key
Participants []string `json:"participants"`
Status string `json:"status"`
CreatedAt time.Time `json:"createdAt"`
CompletedAt *time.Time `json:"completedAt,omitempty"`
}
// TeleportMessage represents a cross-chain teleport message
type TeleportMessage struct {
ID string `json:"id"`
SourceChain string `json:"sourceChain"`
DestChain string `json:"destChain"`
Sender string `json:"sender"`
Receiver string `json:"receiver"`
Payload json.RawMessage `json:"payload"`
Nonce uint64 `json:"nonce"`
SessionID string `json:"sessionId,omitempty"`
Status string `json:"status"`
CreatedAt time.Time `json:"createdAt"`
SignedAt *time.Time `json:"signedAt,omitempty"`
DeliveredAt *time.Time `json:"deliveredAt,omitempty"`
}
// VertexData contains T-Chain specific vertex data
type VertexData struct {
Session *SigningSession `json:"session,omitempty"`
KeyShare *KeyShare `json:"keyShare,omitempty"`
KeyGen *KeyGeneration `json:"keyGen,omitempty"`
Message *TeleportMessage `json:"message,omitempty"`
}
// Adapter implements dag.Adapter for T-Chain MPC operations
type Adapter struct {
rpcEndpoint string
httpClient *http.Client
}
// Compile-time interface compliance check
var _ dag.Adapter = (*Adapter)(nil)
// NewAdapter creates a new T-Chain adapter
func New(rpcEndpoint string) *Adapter {
if rpcEndpoint == "" {
rpcEndpoint = DefaultRPCEndpoint
}
return &Adapter{
rpcEndpoint: rpcEndpoint,
httpClient: &http.Client{Timeout: 30 * time.Second},
}
}
// DefaultConfig returns default configuration for T-Chain indexer
func DefaultConfig() dag.Config {
return dag.Config{
ChainType: dag.ChainT,
ChainName: "T-Chain (Teleport)",
RPCEndpoint: DefaultRPCEndpoint,
RPCMethod: "tvm",
DatabaseURL: DefaultDatabaseURL,
HTTPPort: DefaultHTTPPort,
PollInterval: 2 * time.Second,
}
}
// ParseVertex parses T-Chain vertex from RPC response
func (a *Adapter) ParseVertex(data json.RawMessage) (*dag.Vertex, error) {
var raw struct {
ID string `json:"id"`
Type string `json:"type"`
ParentIDs []string `json:"parentIds"`
Height uint64 `json:"height"`
Epoch uint32 `json:"epoch"`
Timestamp int64 `json:"timestamp"`
Status string `json:"status"`
Data json.RawMessage `json:"data"`
// MPC-specific fields
SessionID string `json:"sessionId,omitempty"`
KeyGenID string `json:"keyGenId,omitempty"`
MessageID string `json:"messageId,omitempty"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, fmt.Errorf("parse vertex: %w", err)
}
v := &dag.Vertex{
ID: raw.ID,
Type: raw.Type,
ParentIDs: raw.ParentIDs,
Height: raw.Height,
Epoch: raw.Epoch,
Timestamp: time.Unix(raw.Timestamp, 0),
Status: dag.Status(raw.Status),
Data: raw.Data,
Metadata: make(map[string]interface{}),
}
// Determine vertex type from content
if v.Type == "" {
v.Type = a.inferVertexType(raw.Data)
}
// Add MPC-specific metadata
if raw.SessionID != "" {
v.Metadata["sessionId"] = raw.SessionID
}
if raw.KeyGenID != "" {
v.Metadata["keyGenId"] = raw.KeyGenID
}
if raw.MessageID != "" {
v.Metadata["messageId"] = raw.MessageID
}
return v, nil
}
// inferVertexType determines vertex type from data content
func (a *Adapter) inferVertexType(data json.RawMessage) string {
var probe map[string]interface{}
if err := json.Unmarshal(data, &probe); err != nil {
return "unknown"
}
if _, ok := probe["threshold"]; ok {
if _, ok := probe["messageHash"]; ok {
return "signing_session"
}
if _, ok := probe["publicKey"]; ok {
return "key_generation"
}
}
if _, ok := probe["shareIndex"]; ok {
return "key_share"
}
if _, ok := probe["sourceChain"]; ok {
return "teleport_message"
}
return "unknown"
}
// GetRecentVertices fetches recent vertices via RPC
func (a *Adapter) GetRecentVertices(ctx context.Context, limit int) ([]json.RawMessage, error) {
req := map[string]interface{}{
"jsonrpc": "2.0",
"id": 1,
"method": "tvm.getRecentVertices",
"params": map[string]interface{}{
"limit": limit,
},
}
resp, err := a.rpcCall(ctx, req)
if err != nil {
return nil, err
}
var result struct {
Vertices []json.RawMessage `json:"vertices"`
}
if err := json.Unmarshal(resp, &result); err != nil {
return nil, fmt.Errorf("parse vertices: %w", err)
}
return result.Vertices, nil
}
// GetVertexByID fetches a specific vertex
func (a *Adapter) GetVertexByID(ctx context.Context, id string) (json.RawMessage, error) {
req := map[string]interface{}{
"jsonrpc": "2.0",
"id": 1,
"method": "tvm.getVertex",
"params": map[string]interface{}{
"id": id,
},
}
resp, err := a.rpcCall(ctx, req)
if err != nil {
return nil, err
}
var result struct {
Vertex json.RawMessage `json:"vertex"`
}
if err := json.Unmarshal(resp, &result); err != nil {
return nil, fmt.Errorf("parse vertex: %w", err)
}
return result.Vertex, nil
}
// GetSigningSession fetches a signing session by ID
func (a *Adapter) GetSigningSession(ctx context.Context, sessionID string) (*SigningSession, error) {
req := map[string]interface{}{
"jsonrpc": "2.0",
"id": 1,
"method": "tvm.getSigningSession",
"params": map[string]interface{}{
"sessionId": sessionID,
},
}
resp, err := a.rpcCall(ctx, req)
if err != nil {
return nil, err
}
var session SigningSession
if err := json.Unmarshal(resp, &session); err != nil {
return nil, fmt.Errorf("parse session: %w", err)
}
return &session, nil
}
// GetKeyGeneration fetches a key generation ceremony by ID
func (a *Adapter) GetKeyGeneration(ctx context.Context, keyGenID string) (*KeyGeneration, error) {
req := map[string]interface{}{
"jsonrpc": "2.0",
"id": 1,
"method": "tvm.getKeyGeneration",
"params": map[string]interface{}{
"keyGenId": keyGenID,
},
}
resp, err := a.rpcCall(ctx, req)
if err != nil {
return nil, err
}
var keyGen KeyGeneration
if err := json.Unmarshal(resp, &keyGen); err != nil {
return nil, fmt.Errorf("parse keygen: %w", err)
}
return &keyGen, nil
}
// GetActiveKeyShares fetches active key shares for a public key
func (a *Adapter) GetActiveKeyShares(ctx context.Context, publicKey string) ([]KeyShare, error) {
req := map[string]interface{}{
"jsonrpc": "2.0",
"id": 1,
"method": "tvm.getActiveKeyShares",
"params": map[string]interface{}{
"publicKey": publicKey,
},
}
resp, err := a.rpcCall(ctx, req)
if err != nil {
return nil, err
}
var result struct {
KeyShares []KeyShare `json:"keyShares"`
}
if err := json.Unmarshal(resp, &result); err != nil {
return nil, fmt.Errorf("parse key shares: %w", err)
}
return result.KeyShares, nil
}
// InitSchema creates T-Chain specific database tables
func (a *Adapter) InitSchema(db *sql.DB) error {
schema := `
-- Signing sessions table
CREATE TABLE IF NOT EXISTS tchain_sessions (
id TEXT PRIMARY KEY,
threshold INT NOT NULL,
total_shares INT NOT NULL,
message_hash TEXT NOT NULL,
participants JSONB DEFAULT '[]',
signatures JSONB DEFAULT '[]',
final_sig TEXT,
status TEXT DEFAULT 'pending',
source_chain TEXT,
dest_chain TEXT,
created_at TIMESTAMPTZ DEFAULT NOW(),
completed_at TIMESTAMPTZ,
expires_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_tchain_sessions_status ON tchain_sessions(status);
CREATE INDEX IF NOT EXISTS idx_tchain_sessions_message ON tchain_sessions(message_hash);
CREATE INDEX IF NOT EXISTS idx_tchain_sessions_created ON tchain_sessions(created_at DESC);
-- Key shares table
CREATE TABLE IF NOT EXISTS tchain_key_shares (
id TEXT PRIMARY KEY,
public_key TEXT NOT NULL,
share_index INT NOT NULL,
node_id TEXT NOT NULL,
keygen_id TEXT NOT NULL,
status TEXT DEFAULT 'pending',
created_at TIMESTAMPTZ DEFAULT NOW(),
rotated_at TIMESTAMPTZ,
expires_at TIMESTAMPTZ
);
CREATE INDEX IF NOT EXISTS idx_tchain_key_shares_pubkey ON tchain_key_shares(public_key);
CREATE INDEX IF NOT EXISTS idx_tchain_key_shares_node ON tchain_key_shares(node_id);
CREATE INDEX IF NOT EXISTS idx_tchain_key_shares_status ON tchain_key_shares(status);
-- Key generation ceremonies table
CREATE TABLE IF NOT EXISTS tchain_key_generations (
id TEXT PRIMARY KEY,
threshold INT NOT NULL,
total_shares INT NOT NULL,
public_key TEXT,
participants JSONB DEFAULT '[]',
status TEXT DEFAULT 'pending',
created_at TIMESTAMPTZ DEFAULT NOW(),
completed_at TIMESTAMPTZ
);
CREATE INDEX IF NOT EXISTS idx_tchain_key_gen_status ON tchain_key_generations(status);
CREATE INDEX IF NOT EXISTS idx_tchain_key_gen_pubkey ON tchain_key_generations(public_key);
-- Teleport messages table
CREATE TABLE IF NOT EXISTS tchain_messages (
id TEXT PRIMARY KEY,
source_chain TEXT NOT NULL,
dest_chain TEXT NOT NULL,
sender TEXT NOT NULL,
receiver TEXT NOT NULL,
payload JSONB,
nonce BIGINT NOT NULL,
session_id TEXT,
status TEXT DEFAULT 'pending',
created_at TIMESTAMPTZ DEFAULT NOW(),
signed_at TIMESTAMPTZ,
delivered_at TIMESTAMPTZ
);
CREATE INDEX IF NOT EXISTS idx_tchain_messages_source ON tchain_messages(source_chain);
CREATE INDEX IF NOT EXISTS idx_tchain_messages_dest ON tchain_messages(dest_chain);
CREATE INDEX IF NOT EXISTS idx_tchain_messages_status ON tchain_messages(status);
CREATE INDEX IF NOT EXISTS idx_tchain_messages_session ON tchain_messages(session_id);
-- MPC statistics table
CREATE TABLE IF NOT EXISTS tchain_mpc_stats (
id INT PRIMARY KEY DEFAULT 1,
total_sessions BIGINT DEFAULT 0,
active_sessions BIGINT DEFAULT 0,
completed_sessions BIGINT DEFAULT 0,
failed_sessions BIGINT DEFAULT 0,
total_key_shares BIGINT DEFAULT 0,
active_key_shares BIGINT DEFAULT 0,
total_key_gens BIGINT DEFAULT 0,
total_messages BIGINT DEFAULT 0,
pending_messages BIGINT DEFAULT 0,
delivered_messages BIGINT DEFAULT 0,
updated_at TIMESTAMPTZ DEFAULT NOW()
);
INSERT INTO tchain_mpc_stats (id) VALUES (1) ON CONFLICT DO NOTHING;
`
if _, err := db.Exec(schema); err != nil {
return fmt.Errorf("init tchain schema: %w", err)
}
return nil
}
// GetStats returns T-Chain specific statistics
func (a *Adapter) GetStats(ctx context.Context, db *sql.DB) (map[string]interface{}, error) {
stats := make(map[string]interface{})
// Get MPC stats
var totalSessions, activeSessions, completedSessions, failedSessions int64
var totalKeyShares, activeKeyShares, totalKeyGens int64
var totalMessages, pendingMessages, deliveredMessages int64
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tchain_sessions").Scan(&totalSessions)
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tchain_sessions WHERE status='active'").Scan(&activeSessions)
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tchain_sessions WHERE status='completed'").Scan(&completedSessions)
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tchain_sessions WHERE status='failed'").Scan(&failedSessions)
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tchain_key_shares").Scan(&totalKeyShares)
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tchain_key_shares WHERE status='active'").Scan(&activeKeyShares)
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tchain_key_generations").Scan(&totalKeyGens)
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tchain_messages").Scan(&totalMessages)
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tchain_messages WHERE status='pending'").Scan(&pendingMessages)
db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tchain_messages WHERE status='delivered'").Scan(&deliveredMessages)
stats["mpc_stats"] = map[string]interface{}{
"total_sessions": totalSessions,
"active_sessions": activeSessions,
"completed_sessions": completedSessions,
"failed_sessions": failedSessions,
}
stats["key_stats"] = map[string]interface{}{
"total_key_shares": totalKeyShares,
"active_key_shares": activeKeyShares,
"total_key_gens": totalKeyGens,
}
stats["message_stats"] = map[string]interface{}{
"total_messages": totalMessages,
"pending_messages": pendingMessages,
"delivered_messages": deliveredMessages,
}
// Calculate signing success rate
if totalSessions > 0 {
stats["signing_success_rate"] = float64(completedSessions) / float64(totalSessions) * 100.0
}
// Get average threshold
var avgThreshold float64
db.QueryRowContext(ctx, "SELECT COALESCE(AVG(threshold), 0) FROM tchain_key_generations WHERE status='completed'").Scan(&avgThreshold)
stats["avg_threshold"] = avgThreshold
return stats, nil
}
// StoreSession stores a signing session
func (a *Adapter) StoreSession(ctx context.Context, db *sql.DB, s *SigningSession) error {
participants, _ := json.Marshal(s.Participants)
signatures, _ := json.Marshal(s.Signatures)
_, err := db.ExecContext(ctx, `
INSERT INTO tchain_sessions (id, threshold, total_shares, message_hash, participants, signatures, final_sig, status, source_chain, dest_chain, created_at, completed_at, expires_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
ON CONFLICT (id) DO UPDATE SET
signatures = EXCLUDED.signatures,
final_sig = EXCLUDED.final_sig,
status = EXCLUDED.status,
completed_at = EXCLUDED.completed_at
`, s.ID, s.Threshold, s.TotalShares, s.MessageHash, participants, signatures, s.FinalSig, s.Status, s.SourceChain, s.DestChain, s.CreatedAt, s.CompletedAt, s.ExpiresAt)
return err
}
// StoreKeyShare stores a key share
func (a *Adapter) StoreKeyShare(ctx context.Context, db *sql.DB, k *KeyShare) error {
_, err := db.ExecContext(ctx, `
INSERT INTO tchain_key_shares (id, public_key, share_index, node_id, keygen_id, status, created_at, rotated_at, expires_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (id) DO UPDATE SET
status = EXCLUDED.status,
rotated_at = EXCLUDED.rotated_at
`, k.ID, k.PublicKey, k.ShareIndex, k.NodeID, k.KeyGenID, k.Status, k.CreatedAt, k.RotatedAt, k.ExpiresAt)
return err
}
// StoreKeyGeneration stores a key generation ceremony
func (a *Adapter) StoreKeyGeneration(ctx context.Context, db *sql.DB, kg *KeyGeneration) error {
participants, _ := json.Marshal(kg.Participants)
_, err := db.ExecContext(ctx, `
INSERT INTO tchain_key_generations (id, threshold, total_shares, public_key, participants, status, created_at, completed_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT (id) DO UPDATE SET
public_key = EXCLUDED.public_key,
status = EXCLUDED.status,
completed_at = EXCLUDED.completed_at
`, kg.ID, kg.Threshold, kg.TotalShares, kg.PublicKey, participants, kg.Status, kg.CreatedAt, kg.CompletedAt)
return err
}
// StoreMessage stores a teleport message
func (a *Adapter) StoreMessage(ctx context.Context, db *sql.DB, m *TeleportMessage) error {
_, err := db.ExecContext(ctx, `
INSERT INTO tchain_messages (id, source_chain, dest_chain, sender, receiver, payload, nonce, session_id, status, created_at, signed_at, delivered_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
ON CONFLICT (id) DO UPDATE SET
session_id = EXCLUDED.session_id,
status = EXCLUDED.status,
signed_at = EXCLUDED.signed_at,
delivered_at = EXCLUDED.delivered_at
`, m.ID, m.SourceChain, m.DestChain, m.Sender, m.Receiver, m.Payload, m.Nonce, m.SessionID, m.Status, m.CreatedAt, m.SignedAt, m.DeliveredAt)
return err
}
// rpcCall makes an RPC call to T-Chain
func (a *Adapter) rpcCall(ctx context.Context, req map[string]interface{}) (json.RawMessage, error) {
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("marshal request: %w", err)
}
httpReq, err := http.NewRequestWithContext(ctx, "POST", a.rpcEndpoint, bytes.NewReader(body))
if err != nil {
return nil, fmt.Errorf("create request: %w", err)
}
httpReq.Header.Set("Content-Type", "application/json")
resp, err := a.httpClient.Do(httpReq)
if err != nil {
return nil, fmt.Errorf("rpc call: %w", err)
}
defer resp.Body.Close()
var rpcResp struct {
Result json.RawMessage `json:"result"`
Error *struct {
Code int `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
if err := json.NewDecoder(resp.Body).Decode(&rpcResp); err != nil {
return nil, fmt.Errorf("decode response: %w", err)
}
if rpcResp.Error != nil {
return nil, fmt.Errorf("rpc error %d: %s", rpcResp.Error.Code, rpcResp.Error.Message)
}
return rpcResp.Result, nil
}
// ValidateSignature validates a threshold signature
func ValidateSignature(publicKey, message, signature string) (bool, error) {
// Decode hex strings
pubKeyBytes, err := hex.DecodeString(publicKey)
if err != nil {
return false, fmt.Errorf("decode public key: %w", err)
}
msgBytes, err := hex.DecodeString(message)
if err != nil {
return false, fmt.Errorf("decode message: %w", err)
}
sigBytes, err := hex.DecodeString(signature)
if err != nil {
return false, fmt.Errorf("decode signature: %w", err)
}
// Basic length validation (actual crypto verification would use luxfi/crypto)
if len(pubKeyBytes) < 32 || len(sigBytes) < 64 {
return false, fmt.Errorf("invalid key or signature length")
}
// Placeholder - actual verification uses threshold signature scheme
_ = msgBytes
return true, nil
}
// ComputeThreshold computes required threshold for given parameters
// Uses t-of-n threshold where t = (n * 2 / 3) + 1 for Byzantine fault tolerance
func ComputeThreshold(totalShares uint32) uint32 {
if totalShares < 2 {
return 1
}
return (totalShares * 2 / 3) + 1
}
+467
View File
@@ -0,0 +1,467 @@
// Copyright (c) 2025 Lux Partners Limited
// SPDX-License-Identifier: MIT
// Package xchain provides the X-Chain (Exchange) adapter for the DAG indexer.
// X-Chain handles asset exchange and UTXOs using DAG consensus.
package xchain
import (
"bytes"
"context"
"database/sql"
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/luxfi/indexer/dag"
)
// Adapter implements dag.Adapter for X-Chain
type Adapter struct {
rpcEndpoint string
client *http.Client
}
// New creates a new X-Chain adapter
func New(rpcEndpoint string) *Adapter {
return &Adapter{
rpcEndpoint: rpcEndpoint,
client: &http.Client{Timeout: 30 * time.Second},
}
}
// UTXO represents an unspent transaction output
type UTXO struct {
ID string `json:"id"`
TxID string `json:"txId"`
OutIndex uint32 `json:"outIndex"`
AssetID string `json:"assetId"`
Amount uint64 `json:"amount"`
Address string `json:"address"`
Spent bool `json:"spent"`
SpentBy string `json:"spentBy,omitempty"`
}
// Asset represents an X-Chain asset
type Asset struct {
ID string `json:"id"`
Name string `json:"name"`
Symbol string `json:"symbol"`
Denomination uint8 `json:"denomination"`
TotalSupply uint64 `json:"totalSupply"`
}
// Transaction represents an X-Chain transaction
type Transaction struct {
ID string `json:"id"`
Type string `json:"type"`
Inputs []Input `json:"inputs"`
Outputs []Output `json:"outputs"`
Memo string `json:"memo,omitempty"`
Timestamp int64 `json:"timestamp"`
}
// Input represents a transaction input
type Input struct {
TxID string `json:"txId"`
OutIndex uint32 `json:"outIndex"`
AssetID string `json:"assetId"`
Amount uint64 `json:"amount"`
Address string `json:"address"`
}
// Output represents a transaction output
type Output struct {
AssetID string `json:"assetId"`
Amount uint64 `json:"amount"`
Addresses []string `json:"addresses"`
Threshold uint32 `json:"threshold"`
Locktime uint64 `json:"locktime"`
}
// VertexData holds X-Chain specific vertex data
type VertexData struct {
Transactions []Transaction `json:"transactions"`
ChainID string `json:"chainId"`
}
// rpcRequest for JSON-RPC calls
type rpcRequest struct {
JSONRPC string `json:"jsonrpc"`
ID int `json:"id"`
Method string `json:"method"`
Params interface{} `json:"params,omitempty"`
}
// rpcResponse for JSON-RPC responses
type rpcResponse struct {
JSONRPC string `json:"jsonrpc"`
ID int `json:"id"`
Result json.RawMessage `json:"result,omitempty"`
Error *rpcError `json:"error,omitempty"`
}
type rpcError struct {
Code int `json:"code"`
Message string `json:"message"`
}
// callRPC makes a JSON-RPC call to the X-Chain
func (a *Adapter) callRPC(ctx context.Context, method string, params interface{}) (json.RawMessage, error) {
req := rpcRequest{
JSONRPC: "2.0",
ID: 1,
Method: method,
Params: params,
}
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("marshal request: %w", err)
}
httpReq, err := http.NewRequestWithContext(ctx, "POST", a.rpcEndpoint, bytes.NewReader(body))
if err != nil {
return nil, fmt.Errorf("create request: %w", err)
}
httpReq.Header.Set("Content-Type", "application/json")
resp, err := a.client.Do(httpReq)
if err != nil {
return nil, fmt.Errorf("rpc call: %w", err)
}
defer resp.Body.Close()
var rpcResp rpcResponse
if err := json.NewDecoder(resp.Body).Decode(&rpcResp); err != nil {
return nil, fmt.Errorf("decode response: %w", err)
}
if rpcResp.Error != nil {
return nil, fmt.Errorf("rpc error %d: %s", rpcResp.Error.Code, rpcResp.Error.Message)
}
return rpcResp.Result, nil
}
// ParseVertex parses X-Chain vertex data into a dag.Vertex
func (a *Adapter) ParseVertex(data json.RawMessage) (*dag.Vertex, error) {
var raw struct {
ID string `json:"id"`
ParentIDs []string `json:"parentIDs"`
Height uint64 `json:"height"`
Epoch uint32 `json:"epoch"`
TxIDs []string `json:"txs"`
Timestamp int64 `json:"timestamp"`
Status string `json:"status"`
ChainID string `json:"chainID"`
Bytes json.RawMessage `json:"bytes,omitempty"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, fmt.Errorf("unmarshal vertex: %w", err)
}
status := dag.StatusPending
switch raw.Status {
case "Accepted":
status = dag.StatusAccepted
case "Rejected":
status = dag.StatusRejected
}
ts := time.Unix(raw.Timestamp, 0)
if raw.Timestamp == 0 {
ts = time.Now()
}
return &dag.Vertex{
ID: raw.ID,
Type: "xchain_vertex",
ParentIDs: raw.ParentIDs,
Height: raw.Height,
Epoch: raw.Epoch,
TxIDs: raw.TxIDs,
Timestamp: ts,
Status: status,
Data: data,
Metadata: map[string]interface{}{
"chainId": raw.ChainID,
},
}, nil
}
// GetRecentVertices fetches recent vertices from the X-Chain
func (a *Adapter) GetRecentVertices(ctx context.Context, limit int) ([]json.RawMessage, error) {
params := map[string]interface{}{
"limit": limit,
}
result, err := a.callRPC(ctx, "xvm.getRecentVertices", params)
if err != nil {
return nil, err
}
var resp struct {
Vertices []json.RawMessage `json:"vertices"`
}
if err := json.Unmarshal(result, &resp); err != nil {
return nil, fmt.Errorf("unmarshal vertices: %w", err)
}
return resp.Vertices, nil
}
// GetVertexByID fetches a specific vertex by ID
func (a *Adapter) GetVertexByID(ctx context.Context, id string) (json.RawMessage, error) {
params := map[string]interface{}{
"vertexID": id,
}
result, err := a.callRPC(ctx, "xvm.getVertex", params)
if err != nil {
return nil, err
}
var resp struct {
Vertex json.RawMessage `json:"vertex"`
}
if err := json.Unmarshal(result, &resp); err != nil {
return nil, fmt.Errorf("unmarshal vertex: %w", err)
}
return resp.Vertex, nil
}
// InitSchema creates X-Chain specific database tables
func (a *Adapter) InitSchema(db *sql.DB) error {
schema := `
-- X-Chain assets
CREATE TABLE IF NOT EXISTS xchain_assets (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
symbol TEXT NOT NULL,
denomination SMALLINT DEFAULT 0,
total_supply BIGINT DEFAULT 0,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_xchain_assets_symbol ON xchain_assets(symbol);
-- X-Chain UTXOs
CREATE TABLE IF NOT EXISTS xchain_utxos (
id TEXT PRIMARY KEY,
tx_id TEXT NOT NULL,
out_index INT NOT NULL,
asset_id TEXT NOT NULL REFERENCES xchain_assets(id),
amount BIGINT NOT NULL,
address TEXT NOT NULL,
spent BOOLEAN DEFAULT FALSE,
spent_by TEXT,
created_at TIMESTAMPTZ DEFAULT NOW(),
spent_at TIMESTAMPTZ
);
CREATE INDEX IF NOT EXISTS idx_xchain_utxos_address ON xchain_utxos(address);
CREATE INDEX IF NOT EXISTS idx_xchain_utxos_asset ON xchain_utxos(asset_id);
CREATE INDEX IF NOT EXISTS idx_xchain_utxos_unspent ON xchain_utxos(spent) WHERE NOT spent;
CREATE INDEX IF NOT EXISTS idx_xchain_utxos_tx ON xchain_utxos(tx_id);
-- X-Chain transactions
CREATE TABLE IF NOT EXISTS xchain_transactions (
id TEXT PRIMARY KEY,
vertex_id TEXT,
type TEXT NOT NULL,
memo TEXT,
timestamp TIMESTAMPTZ NOT NULL,
inputs JSONB DEFAULT '[]',
outputs JSONB DEFAULT '[]',
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_xchain_transactions_vertex ON xchain_transactions(vertex_id);
CREATE INDEX IF NOT EXISTS idx_xchain_transactions_type ON xchain_transactions(type);
CREATE INDEX IF NOT EXISTS idx_xchain_transactions_timestamp ON xchain_transactions(timestamp DESC);
-- X-Chain address balances (materialized view alternative)
CREATE TABLE IF NOT EXISTS xchain_balances (
address TEXT NOT NULL,
asset_id TEXT NOT NULL REFERENCES xchain_assets(id),
balance BIGINT DEFAULT 0,
utxo_count INT DEFAULT 0,
updated_at TIMESTAMPTZ DEFAULT NOW(),
PRIMARY KEY (address, asset_id)
);
CREATE INDEX IF NOT EXISTS idx_xchain_balances_address ON xchain_balances(address);
CREATE INDEX IF NOT EXISTS idx_xchain_balances_asset ON xchain_balances(asset_id);
-- X-Chain statistics
CREATE TABLE IF NOT EXISTS xchain_chain_stats (
id INT PRIMARY KEY DEFAULT 1,
total_assets INT DEFAULT 0,
total_utxos BIGINT DEFAULT 0,
unspent_utxos BIGINT DEFAULT 0,
total_transactions BIGINT DEFAULT 0,
total_addresses BIGINT DEFAULT 0,
updated_at TIMESTAMPTZ DEFAULT NOW()
);
INSERT INTO xchain_chain_stats (id) VALUES (1) ON CONFLICT DO NOTHING;
-- Insert LUX as default asset if not exists
INSERT INTO xchain_assets (id, name, symbol, denomination, total_supply)
VALUES ('LUX', 'Lux', 'LUX', 9, 0)
ON CONFLICT DO NOTHING;
`
if _, err := db.Exec(schema); err != nil {
return fmt.Errorf("xchain schema: %w", err)
}
return nil
}
// GetStats returns X-Chain specific statistics
func (a *Adapter) GetStats(ctx context.Context, db *sql.DB) (map[string]interface{}, error) {
// Update stats first
_, _ = db.ExecContext(ctx, `
UPDATE xchain_chain_stats SET
total_assets = (SELECT COUNT(*) FROM xchain_assets),
total_utxos = (SELECT COUNT(*) FROM xchain_utxos),
unspent_utxos = (SELECT COUNT(*) FROM xchain_utxos WHERE NOT spent),
total_transactions = (SELECT COUNT(*) FROM xchain_transactions),
total_addresses = (SELECT COUNT(DISTINCT address) FROM xchain_utxos),
updated_at = NOW()
WHERE id = 1
`)
var stats struct {
TotalAssets int `json:"total_assets"`
TotalUTXOs int64 `json:"total_utxos"`
UnspentUTXOs int64 `json:"unspent_utxos"`
TotalTransactions int64 `json:"total_transactions"`
TotalAddresses int64 `json:"total_addresses"`
}
err := db.QueryRowContext(ctx, `
SELECT total_assets, total_utxos, unspent_utxos, total_transactions, total_addresses
FROM xchain_chain_stats WHERE id = 1
`).Scan(&stats.TotalAssets, &stats.TotalUTXOs, &stats.UnspentUTXOs,
&stats.TotalTransactions, &stats.TotalAddresses)
if err != nil {
return nil, fmt.Errorf("query stats: %w", err)
}
return map[string]interface{}{
"total_assets": stats.TotalAssets,
"total_utxos": stats.TotalUTXOs,
"unspent_utxos": stats.UnspentUTXOs,
"total_transactions": stats.TotalTransactions,
"total_addresses": stats.TotalAddresses,
}, nil
}
// StoreTransaction stores a transaction and updates UTXOs
func (a *Adapter) StoreTransaction(ctx context.Context, db *sql.DB, vertexID string, tx *Transaction) error {
inputsJSON, _ := json.Marshal(tx.Inputs)
outputsJSON, _ := json.Marshal(tx.Outputs)
// Store transaction
_, err := db.ExecContext(ctx, `
INSERT INTO xchain_transactions (id, vertex_id, type, memo, timestamp, inputs, outputs)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (id) DO NOTHING
`, tx.ID, vertexID, tx.Type, tx.Memo, time.Unix(tx.Timestamp, 0), inputsJSON, outputsJSON)
if err != nil {
return fmt.Errorf("store transaction: %w", err)
}
// Mark inputs as spent
for _, in := range tx.Inputs {
utxoID := fmt.Sprintf("%s:%d", in.TxID, in.OutIndex)
_, _ = db.ExecContext(ctx, `
UPDATE xchain_utxos SET spent = TRUE, spent_by = $1, spent_at = NOW()
WHERE id = $2
`, tx.ID, utxoID)
}
// Create new UTXOs from outputs
for i, out := range tx.Outputs {
if len(out.Addresses) == 0 {
continue
}
utxoID := fmt.Sprintf("%s:%d", tx.ID, i)
_, _ = db.ExecContext(ctx, `
INSERT INTO xchain_utxos (id, tx_id, out_index, asset_id, amount, address)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (id) DO NOTHING
`, utxoID, tx.ID, i, out.AssetID, out.Amount, out.Addresses[0])
}
return nil
}
// StoreAsset stores or updates an asset
func (a *Adapter) StoreAsset(ctx context.Context, db *sql.DB, asset *Asset) error {
_, err := db.ExecContext(ctx, `
INSERT INTO xchain_assets (id, name, symbol, denomination, total_supply)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (id) DO UPDATE SET
name = EXCLUDED.name,
symbol = EXCLUDED.symbol,
total_supply = EXCLUDED.total_supply
`, asset.ID, asset.Name, asset.Symbol, asset.Denomination, asset.TotalSupply)
return err
}
// GetUTXOs fetches UTXOs for an address
func (a *Adapter) GetUTXOs(ctx context.Context, db *sql.DB, address string, assetID string, spentFilter *bool) ([]UTXO, error) {
query := `SELECT id, tx_id, out_index, asset_id, amount, address, spent, spent_by
FROM xchain_utxos WHERE address = $1`
args := []interface{}{address}
if assetID != "" {
query += " AND asset_id = $2"
args = append(args, assetID)
}
if spentFilter != nil {
if len(args) == 1 {
query += " AND spent = $2"
} else {
query += " AND spent = $3"
}
args = append(args, *spentFilter)
}
query += " ORDER BY amount DESC"
rows, err := db.QueryContext(ctx, query, args...)
if err != nil {
return nil, fmt.Errorf("query utxos: %w", err)
}
defer rows.Close()
var utxos []UTXO
for rows.Next() {
var u UTXO
var spentBy sql.NullString
if err := rows.Scan(&u.ID, &u.TxID, &u.OutIndex, &u.AssetID, &u.Amount, &u.Address, &u.Spent, &spentBy); err != nil {
continue
}
u.SpentBy = spentBy.String
utxos = append(utxos, u)
}
return utxos, nil
}
// GetBalance calculates balance for an address
func (a *Adapter) GetBalance(ctx context.Context, db *sql.DB, address string, assetID string) (uint64, error) {
var balance uint64
err := db.QueryRowContext(ctx, `
SELECT COALESCE(SUM(amount), 0) FROM xchain_utxos
WHERE address = $1 AND asset_id = $2 AND NOT spent
`, address, assetID).Scan(&balance)
return balance, err
}
// Verify Adapter implements dag.Adapter interface
var _ dag.Adapter = (*Adapter)(nil)
+607
View File
@@ -0,0 +1,607 @@
// Copyright (c) 2025 Lux Partners Limited
// SPDX-License-Identifier: MIT
// Package zchain provides the Z-Chain (Privacy) adapter for DAG indexing.
// Z-Chain uses DAG consensus for fast finality combined with ZK proofs for privacy.
package zchain
import (
"bytes"
"context"
"database/sql"
"encoding/hex"
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/luxfi/indexer/dag"
)
const (
// DefaultRPCEndpoint for Z-Chain
DefaultRPCEndpoint = "http://localhost:9630/ext/bc/Z/rpc"
// DefaultHTTPPort for Z-Chain indexer API
DefaultHTTPPort = 4400
// DefaultDatabase for Z-Chain explorer
DefaultDatabase = "explorer_zchain"
// RPCMethod prefix for Z-Chain
RPCMethod = "zvm"
)
// ZKTransactionType identifies privacy transaction types
type ZKTransactionType string
const (
TxShieldedTransfer ZKTransactionType = "shielded_transfer"
TxShield ZKTransactionType = "shield" // Deposit: transparent -> shielded
TxUnshield ZKTransactionType = "unshield" // Withdraw: shielded -> transparent
TxJoinSplit ZKTransactionType = "joinsplit"
TxMint ZKTransactionType = "mint"
TxBurn ZKTransactionType = "burn"
)
// ProofType identifies ZK proof systems
type ProofType string
const (
ProofGroth16 ProofType = "groth16"
ProofPlonk ProofType = "plonk"
ProofSTARK ProofType = "stark"
ProofBullet ProofType = "bulletproof"
ProofHalo2 ProofType = "halo2"
ProofFRI ProofType = "fri" // Fast Reed-Solomon IOP
)
// ZKProof represents a zero-knowledge proof
type ZKProof struct {
Type ProofType `json:"type"`
Data string `json:"data"` // Hex-encoded proof bytes
PublicInputs []string `json:"publicInputs,omitempty"`
VerifyingKey string `json:"verifyingKey,omitempty"`
}
// Nullifier represents a spent note marker (prevents double-spending)
type Nullifier struct {
Hash string `json:"hash"`
TxID string `json:"txId"`
Index int `json:"index"`
SpentAt int64 `json:"spentAt,omitempty"` // Vertex height
}
// Commitment represents a shielded note commitment
type Commitment struct {
Hash string `json:"hash"`
TxID string `json:"txId"`
Index int `json:"index"`
CreatedAt int64 `json:"createdAt,omitempty"` // Vertex height
Spent bool `json:"spent"`
}
// ShieldedTransfer represents a private transfer
type ShieldedTransfer struct {
TxID string `json:"txId"`
Type ZKTransactionType `json:"type"`
Proof ZKProof `json:"proof"`
Nullifiers []Nullifier `json:"nullifiers"`
Commitments []Commitment `json:"commitments"`
ValueBalance int64 `json:"valueBalance,omitempty"` // Net flow for shield/unshield
Fee uint64 `json:"fee"`
Memo string `json:"memo,omitempty"` // Encrypted memo
Timestamp time.Time `json:"timestamp"`
}
// VertexData holds Z-Chain specific vertex payload
type VertexData struct {
Transfers []ShieldedTransfer `json:"transfers,omitempty"`
MerkleRoot string `json:"merkleRoot"` // Commitment tree root
NullifierRoot string `json:"nullifierRoot"` // Nullifier set root
Epoch uint32 `json:"epoch"`
Proposer string `json:"proposer,omitempty"`
}
// RPCRequest for JSON-RPC calls
type RPCRequest struct {
JSONRPC string `json:"jsonrpc"`
ID int `json:"id"`
Method string `json:"method"`
Params []interface{} `json:"params,omitempty"`
}
// RPCResponse from JSON-RPC calls
type RPCResponse struct {
JSONRPC string `json:"jsonrpc"`
ID int `json:"id"`
Result json.RawMessage `json:"result,omitempty"`
Error *RPCError `json:"error,omitempty"`
}
// RPCError in JSON-RPC response
type RPCError struct {
Code int `json:"code"`
Message string `json:"message"`
}
// Adapter implements dag.Adapter for Z-Chain
type Adapter struct {
rpcEndpoint string
httpClient *http.Client
}
// New creates a new Z-Chain adapter
func New(rpcEndpoint string) *Adapter {
if rpcEndpoint == "" {
rpcEndpoint = DefaultRPCEndpoint
}
return &Adapter{
rpcEndpoint: rpcEndpoint,
httpClient: &http.Client{Timeout: 30 * time.Second},
}
}
// ParseVertex parses RPC response into a DAG vertex
func (a *Adapter) ParseVertex(data json.RawMessage) (*dag.Vertex, error) {
var raw struct {
ID string `json:"id"`
ParentIDs []string `json:"parentIds"`
Height uint64 `json:"height"`
Epoch uint32 `json:"epoch"`
Timestamp int64 `json:"timestamp"`
Status string `json:"status"`
MerkleRoot string `json:"merkleRoot"`
NullifierRoot string `json:"nullifierRoot"`
Transfers json.RawMessage `json:"transfers"`
Proposer string `json:"proposer"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, fmt.Errorf("parse vertex: %w", err)
}
// Parse transfers
var transfers []ShieldedTransfer
if len(raw.Transfers) > 0 {
if err := json.Unmarshal(raw.Transfers, &transfers); err != nil {
// Transfers may be empty or malformed; continue
transfers = nil
}
}
// Build vertex data
vd := VertexData{
Transfers: transfers,
MerkleRoot: raw.MerkleRoot,
NullifierRoot: raw.NullifierRoot,
Epoch: raw.Epoch,
Proposer: raw.Proposer,
}
vdBytes, _ := json.Marshal(vd)
// Extract transaction IDs
var txIDs []string
for _, t := range transfers {
txIDs = append(txIDs, t.TxID)
}
// Map status
status := dag.StatusPending
switch raw.Status {
case "accepted", "Accepted":
status = dag.StatusAccepted
case "rejected", "Rejected":
status = dag.StatusRejected
}
return &dag.Vertex{
ID: raw.ID,
Type: "zk_vertex",
ParentIDs: raw.ParentIDs,
Height: raw.Height,
Epoch: raw.Epoch,
TxIDs: txIDs,
Timestamp: time.Unix(raw.Timestamp, 0),
Status: status,
Data: vdBytes,
Metadata: map[string]interface{}{
"merkleRoot": raw.MerkleRoot,
"nullifierRoot": raw.NullifierRoot,
"transferCount": len(transfers),
},
}, nil
}
// GetRecentVertices fetches recent vertices from RPC
func (a *Adapter) GetRecentVertices(ctx context.Context, limit int) ([]json.RawMessage, error) {
req := RPCRequest{
JSONRPC: "2.0",
ID: 1,
Method: fmt.Sprintf("%s.getRecentVertices", RPCMethod),
Params: []interface{}{limit},
}
resp, err := a.rpcCall(ctx, req)
if err != nil {
return nil, err
}
var vertices []json.RawMessage
if err := json.Unmarshal(resp.Result, &vertices); err != nil {
return nil, fmt.Errorf("parse vertices: %w", err)
}
return vertices, nil
}
// GetVertexByID fetches a specific vertex
func (a *Adapter) GetVertexByID(ctx context.Context, id string) (json.RawMessage, error) {
req := RPCRequest{
JSONRPC: "2.0",
ID: 1,
Method: fmt.Sprintf("%s.getVertex", RPCMethod),
Params: []interface{}{id},
}
resp, err := a.rpcCall(ctx, req)
if err != nil {
return nil, err
}
return resp.Result, nil
}
// GetNullifier fetches nullifier status
func (a *Adapter) GetNullifier(ctx context.Context, hash string) (*Nullifier, error) {
req := RPCRequest{
JSONRPC: "2.0",
ID: 1,
Method: fmt.Sprintf("%s.getNullifier", RPCMethod),
Params: []interface{}{hash},
}
resp, err := a.rpcCall(ctx, req)
if err != nil {
return nil, err
}
var n Nullifier
if err := json.Unmarshal(resp.Result, &n); err != nil {
return nil, fmt.Errorf("parse nullifier: %w", err)
}
return &n, nil
}
// GetCommitment fetches commitment status
func (a *Adapter) GetCommitment(ctx context.Context, hash string) (*Commitment, error) {
req := RPCRequest{
JSONRPC: "2.0",
ID: 1,
Method: fmt.Sprintf("%s.getCommitment", RPCMethod),
Params: []interface{}{hash},
}
resp, err := a.rpcCall(ctx, req)
if err != nil {
return nil, err
}
var c Commitment
if err := json.Unmarshal(resp.Result, &c); err != nil {
return nil, fmt.Errorf("parse commitment: %w", err)
}
return &c, nil
}
// GetMerkleRoot fetches the current commitment tree root
func (a *Adapter) GetMerkleRoot(ctx context.Context) (string, error) {
req := RPCRequest{
JSONRPC: "2.0",
ID: 1,
Method: fmt.Sprintf("%s.getMerkleRoot", RPCMethod),
}
resp, err := a.rpcCall(ctx, req)
if err != nil {
return "", err
}
var root string
if err := json.Unmarshal(resp.Result, &root); err != nil {
return "", fmt.Errorf("parse merkle root: %w", err)
}
return root, nil
}
// VerifyProof verifies a ZK proof on-chain
func (a *Adapter) VerifyProof(ctx context.Context, proof ZKProof) (bool, error) {
req := RPCRequest{
JSONRPC: "2.0",
ID: 1,
Method: fmt.Sprintf("%s.verifyProof", RPCMethod),
Params: []interface{}{proof},
}
resp, err := a.rpcCall(ctx, req)
if err != nil {
return false, err
}
var valid bool
if err := json.Unmarshal(resp.Result, &valid); err != nil {
return false, fmt.Errorf("parse verify result: %w", err)
}
return valid, nil
}
// InitSchema creates Z-Chain specific database tables
func (a *Adapter) InitSchema(db *sql.DB) error {
schema := `
-- Nullifiers: spent note markers
CREATE TABLE IF NOT EXISTS zchain_nullifiers (
hash TEXT PRIMARY KEY,
tx_id TEXT NOT NULL,
idx INT NOT NULL,
spent_at BIGINT,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_zchain_nullifiers_tx ON zchain_nullifiers(tx_id);
-- Commitments: shielded note commitments
CREATE TABLE IF NOT EXISTS zchain_commitments (
hash TEXT PRIMARY KEY,
tx_id TEXT NOT NULL,
idx INT NOT NULL,
created_at BIGINT,
spent BOOLEAN DEFAULT FALSE,
updated_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_zchain_commitments_tx ON zchain_commitments(tx_id);
CREATE INDEX IF NOT EXISTS idx_zchain_commitments_spent ON zchain_commitments(spent);
-- Shielded transfers
CREATE TABLE IF NOT EXISTS zchain_transfers (
tx_id TEXT PRIMARY KEY,
type TEXT NOT NULL,
proof_type TEXT NOT NULL,
proof_data TEXT NOT NULL,
value_balance BIGINT DEFAULT 0,
fee BIGINT NOT NULL,
memo TEXT,
timestamp TIMESTAMPTZ NOT NULL,
vertex_id TEXT,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_zchain_transfers_type ON zchain_transfers(type);
CREATE INDEX IF NOT EXISTS idx_zchain_transfers_timestamp ON zchain_transfers(timestamp DESC);
CREATE INDEX IF NOT EXISTS idx_zchain_transfers_vertex ON zchain_transfers(vertex_id);
-- ZK proofs archive
CREATE TABLE IF NOT EXISTS zchain_proofs (
id SERIAL PRIMARY KEY,
tx_id TEXT NOT NULL,
proof_type TEXT NOT NULL,
proof_data TEXT NOT NULL,
public_inputs JSONB,
verifying_key TEXT,
verified BOOLEAN DEFAULT FALSE,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_zchain_proofs_tx ON zchain_proofs(tx_id);
CREATE INDEX IF NOT EXISTS idx_zchain_proofs_type ON zchain_proofs(proof_type);
-- Merkle tree state (commitment accumulator)
CREATE TABLE IF NOT EXISTS zchain_merkle_state (
id INT PRIMARY KEY DEFAULT 1,
root TEXT NOT NULL,
height BIGINT NOT NULL,
leaf_count BIGINT NOT NULL,
updated_at TIMESTAMPTZ DEFAULT NOW()
);
INSERT INTO zchain_merkle_state (id, root, height, leaf_count)
VALUES (1, '', 0, 0) ON CONFLICT DO NOTHING;
-- Extended stats
CREATE TABLE IF NOT EXISTS zchain_extended_stats (
id INT PRIMARY KEY DEFAULT 1,
total_nullifiers BIGINT DEFAULT 0,
total_commitments BIGINT DEFAULT 0,
total_transfers BIGINT DEFAULT 0,
shielded_volume BIGINT DEFAULT 0,
shield_count BIGINT DEFAULT 0,
unshield_count BIGINT DEFAULT 0,
updated_at TIMESTAMPTZ DEFAULT NOW()
);
INSERT INTO zchain_extended_stats (id) VALUES (1) ON CONFLICT DO NOTHING;
`
if _, err := db.Exec(schema); err != nil {
return fmt.Errorf("zchain schema: %w", err)
}
return nil
}
// GetStats returns Z-Chain specific statistics
func (a *Adapter) GetStats(ctx context.Context, db *sql.DB) (map[string]interface{}, error) {
stats := make(map[string]interface{})
// Fetch extended stats
var totalNullifiers, totalCommitments, totalTransfers int64
var shieldedVolume, shieldCount, unshieldCount int64
err := db.QueryRowContext(ctx, `
SELECT total_nullifiers, total_commitments, total_transfers,
shielded_volume, shield_count, unshield_count
FROM zchain_extended_stats WHERE id=1
`).Scan(&totalNullifiers, &totalCommitments, &totalTransfers,
&shieldedVolume, &shieldCount, &unshieldCount)
if err != nil && err != sql.ErrNoRows {
return nil, err
}
// Fetch merkle state
var merkleRoot string
var merkleHeight, leafCount int64
_ = db.QueryRowContext(ctx, `
SELECT root, height, leaf_count FROM zchain_merkle_state WHERE id=1
`).Scan(&merkleRoot, &merkleHeight, &leafCount)
// Count unspent commitments
var unspentCommitments int64
_ = db.QueryRowContext(ctx, `
SELECT COUNT(*) FROM zchain_commitments WHERE spent=FALSE
`).Scan(&unspentCommitments)
stats["nullifiers"] = map[string]interface{}{
"total": totalNullifiers,
}
stats["commitments"] = map[string]interface{}{
"total": totalCommitments,
"unspent": unspentCommitments,
}
stats["transfers"] = map[string]interface{}{
"total": totalTransfers,
"shieldedVolume": shieldedVolume,
"shieldCount": shieldCount,
"unshieldCount": unshieldCount,
}
stats["merkle"] = map[string]interface{}{
"root": merkleRoot,
"height": merkleHeight,
"leafCount": leafCount,
}
return stats, nil
}
// StoreTransfer stores a shielded transfer and its components
func (a *Adapter) StoreTransfer(ctx context.Context, db *sql.DB, t ShieldedTransfer, vertexID string) error {
// Store transfer
_, err := db.ExecContext(ctx, `
INSERT INTO zchain_transfers (tx_id, type, proof_type, proof_data, value_balance, fee, memo, timestamp, vertex_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (tx_id) DO NOTHING
`, t.TxID, t.Type, t.Proof.Type, t.Proof.Data, t.ValueBalance, t.Fee, t.Memo, t.Timestamp, vertexID)
if err != nil {
return fmt.Errorf("store transfer: %w", err)
}
// Store proof
publicInputsJSON, _ := json.Marshal(t.Proof.PublicInputs)
_, err = db.ExecContext(ctx, `
INSERT INTO zchain_proofs (tx_id, proof_type, proof_data, public_inputs, verifying_key, verified)
VALUES ($1, $2, $3, $4, $5, TRUE)
`, t.TxID, t.Proof.Type, t.Proof.Data, publicInputsJSON, t.Proof.VerifyingKey)
if err != nil {
return fmt.Errorf("store proof: %w", err)
}
// Store nullifiers
for _, n := range t.Nullifiers {
_, err = db.ExecContext(ctx, `
INSERT INTO zchain_nullifiers (hash, tx_id, idx, spent_at)
VALUES ($1, $2, $3, $4)
ON CONFLICT (hash) DO NOTHING
`, n.Hash, n.TxID, n.Index, n.SpentAt)
if err != nil {
return fmt.Errorf("store nullifier: %w", err)
}
// Mark corresponding commitment as spent
_, _ = db.ExecContext(ctx, `
UPDATE zchain_commitments SET spent=TRUE, updated_at=NOW()
WHERE hash IN (SELECT hash FROM zchain_commitments WHERE spent=FALSE LIMIT 1)
`)
}
// Store commitments
for _, c := range t.Commitments {
_, err = db.ExecContext(ctx, `
INSERT INTO zchain_commitments (hash, tx_id, idx, created_at, spent)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (hash) DO NOTHING
`, c.Hash, c.TxID, c.Index, c.CreatedAt, c.Spent)
if err != nil {
return fmt.Errorf("store commitment: %w", err)
}
}
return nil
}
// UpdateExtendedStats updates Z-Chain specific statistics
func (a *Adapter) UpdateExtendedStats(ctx context.Context, db *sql.DB) error {
_, err := db.ExecContext(ctx, `
UPDATE zchain_extended_stats SET
total_nullifiers = (SELECT COUNT(*) FROM zchain_nullifiers),
total_commitments = (SELECT COUNT(*) FROM zchain_commitments),
total_transfers = (SELECT COUNT(*) FROM zchain_transfers),
shield_count = (SELECT COUNT(*) FROM zchain_transfers WHERE type='shield'),
unshield_count = (SELECT COUNT(*) FROM zchain_transfers WHERE type='unshield'),
shielded_volume = COALESCE((SELECT SUM(ABS(value_balance)) FROM zchain_transfers), 0),
updated_at = NOW()
WHERE id=1
`)
return err
}
// rpcCall executes a JSON-RPC request
func (a *Adapter) rpcCall(ctx context.Context, req RPCRequest) (*RPCResponse, error) {
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("marshal request: %w", err)
}
httpReq, err := http.NewRequestWithContext(ctx, "POST", a.rpcEndpoint, bytes.NewReader(body))
if err != nil {
return nil, fmt.Errorf("create request: %w", err)
}
httpReq.Header.Set("Content-Type", "application/json")
httpResp, err := a.httpClient.Do(httpReq)
if err != nil {
return nil, fmt.Errorf("rpc call: %w", err)
}
defer httpResp.Body.Close()
var resp RPCResponse
if err := json.NewDecoder(httpResp.Body).Decode(&resp); err != nil {
return nil, fmt.Errorf("decode response: %w", err)
}
if resp.Error != nil {
return nil, fmt.Errorf("rpc error %d: %s", resp.Error.Code, resp.Error.Message)
}
return &resp, nil
}
// DefaultConfig returns the default configuration for Z-Chain indexer
func DefaultConfig() dag.Config {
return dag.Config{
ChainType: dag.ChainZ,
ChainName: "Z-Chain (Privacy)",
RPCEndpoint: DefaultRPCEndpoint,
RPCMethod: RPCMethod,
DatabaseURL: fmt.Sprintf("postgres://blockscout:blockscout@localhost:5432/%s?sslmode=disable", DefaultDatabase),
HTTPPort: DefaultHTTPPort,
PollInterval: 5 * time.Second,
}
}
// ValidateNullifierHash validates a nullifier hash format
func ValidateNullifierHash(hash string) error {
if len(hash) != 64 {
return fmt.Errorf("invalid nullifier hash length: expected 64, got %d", len(hash))
}
if _, err := hex.DecodeString(hash); err != nil {
return fmt.Errorf("invalid nullifier hash encoding: %w", err)
}
return nil
}
// ValidateCommitmentHash validates a commitment hash format
func ValidateCommitmentHash(hash string) error {
if len(hash) != 64 {
return fmt.Errorf("invalid commitment hash length: expected 64, got %d", len(hash))
}
if _, err := hex.DecodeString(hash); err != nil {
return fmt.Errorf("invalid commitment hash encoding: %w", err)
}
return nil
}