bitpro가 원문에서 글자만 모은 내용이에요. 표·그림·수식은 빠지거나 깨질 수 있어요. 전체 54쪽 중 12쪽
ocs/data-streams/working-with-multiple-publishers-in-a-shared-stream#tutorial-building-a-multi-publisher-aggregator-app)" tutorial. You will need a `.env` file with at least one private key to act as a publisher, and we will simulate the other addresses.
[**`src/lib/clients.ts`**](https://emre-gitbook.gitbook.io/emre-gitbook-docs/data-streams/working-with-multiple-publishers-in-a-shared-stream#chain-and-client-configuration) (No changes needed from the previous tutorial. We just need `publicClient`.)
**`src/lib/schema.ts`**
```typescript
export const oraclePriceSchema = 'uint256 price, uint64 timestamp'
```
### **The Verification Script**
This script will not publish data. We will assume our two trusted oracles (`PUBLISHER_1_PK` and `PUBLISHER_2_PK` from the previous tutorial) have already published data using the `oraclePriceSchema`.
Our script will:
1. Define a list of `TRUSTED_ORACLES`.
2. Define an `IMPOSTER_ORACLE` (a random address that has *not* published).
3. Create a `verifyPublisher` function that fetches data *only* for a specific publisher address.
4. Run verification for all addresses and show that data is only returned for the correct publishers.
**`src/scripts/verifyOrigin.ts`**
```typescript
import 'dotenv/config'
import { SDK, SchemaDecodedItem } from '@somnia-chain/streams'
import { publicClient } from '../lib/clients' // Assuming you have clients.ts from previous tutorial
import { oraclePriceSchema } from '../lib/schema'
import { Address, createWalletClient, http } from 'viem'
import { privateKeyToAccount } from 'viem/accounts'
// --- Setup: Define our trusted and untrusted addresses ---
function getEnv(key: string): string {
const value = process.env[key]
if (!value) throw new Error(`Missing environment variable: ${key}`)
return value
}
// These are the addresses we trust for this schema.
// We get them from our .env file for this example.
const TRUSTED_ORACLES: Address[] = [
privateKeyToAccount(getEnv('PUBLISHER_1_PK') as `0x${string}`).address,
privateKeyToAccount(getEnv('PUBLISHER_2_PK') as `0x${string}`).address,
]
// This is a random, untrusted address.
const IMPOSTER_ORACLE: Address = '0x1234567890123456789012345678901234567890'
// --- Helper Functions ---
// Helper to decode the oracle data
function decodePriceRecord(row: SchemaDecodedItem[]): { price: bigint, timestamp: number } {
const val = (field: any) => field?.value?.value ?? field?.value ?? ''
return {
price: BigInt(val(row[0])),
timestamp: Number(val(r[1])),
}
}
/**
* Verification Utility
* Fetches data for a *single* publisher to verify its origin.
*/
async function verifyPublisher(sdk: SDK, schemaId: `0x${string}`, publisherAddress: Address) {
console.log(`\n--- Verifying Publisher: ${publisherAddress} ---`)
try {
const data = await sdk.streams.getAllPublisherDataForSchema(schemaId, publisherAddress)
if (!data || data.length === 0) {
console.log('[VERIFIED] No data found for this publisher.')
return
}
const records = (data as SchemaDecodedItem[][]).map(decodePriceRecord)
console.log(`[VERIFIED] Found ${records.length} record(s) cryptographically signed by this publisher:`)
records.forEach(record => {
console.log(` - Price: ${record.price}, Time: ${new Date(record.timestamp).toISOString()}`)
})
} catch (error: any) {
console.error(`Error during verification: ${error.message}`)
}
}
// --- Main Execution ---
async function main() {
const sdk = new SDK({ public: publicClient })
const schemaId = await sdk.streams.computeSchemaId(oraclePriceSchema)
if (!schemaId) throw new Error('Could not compute schemaId')
console.log('Starting Data Provenance Verification...')
console.log(`Schema: oraclePriceSchema (${schemaId})`)
// 1. Verify our trusted oracles
for (const oracleAddress of TRUSTED_ORACLES) {
await verifyPublisher(sdk, schemaId, oracleAddress)
}
// 2. Verify the imposter
// This will securely return NO data, even if the imposter
// published data to the same schemaId under their *own* address.
await verifyPublisher(sdk, schemaId, IMPOSTER_ORACLE)
}
main().catch((e) => {
console.error(e)
process.exit(1)
})
```
### **Expected Output**
To run this, first publish some data (using the script from the previous tutorial, but adapted for `oraclePriceSchema`) from both `PUBLISHER_1_PK` and `PUBLISHER_2_PK`. Then, run the verification script.
```bash
# Add to package.json
"verify": "ts-node src/scripts/verifyOrigin.ts"
# Run it
npm run verify
```
You will see an output similar to this:
```bash
Starting Data Provenance Verification...
Schema: oraclePriceSchema (0x...)
--- Verifying Publisher: 0xPublisher1Address... ---
[VERIFIED] Found 2 record(s) cryptographically signed by this publisher:
- Price: 3200, Time: 2025-10-31T12:30:00.000Z
- Price: 3201, Time: 2025-10-31T12:31:00.000Z
--- Verifying Publisher: 0xPublisher2Address... ---
[VERIFIED] Found 1 record(s) cryptographically signed by this publisher:
- Price: 3199, Time: 2025-10-31T12:30:30.000Z
--- Verifying Publisher: 0x1234567890123456789012345678901234567890 ---
[VERIFIED] No data found for this publisher.
```
## Conclusion: Key Takeaways
* **Provenance is Built-In:** Data provenance in Somnia Streams is not an optional feature; it is a core cryptographic guarantee of the `Streams` smart contract, enforced by `msg.sender`.
* **Verification is Implicit:** You verify data origin every time you perform a read operation with `getAllPublisherDataForSchema` or `getByKey`. The `publisher` address acts as the ultimate verification key.
* **Trust Layer:** This architecture creates a robust trust layer. Your application logic can be certain that any data returned for a specific publisher was, without question, signed and submitted by that publisher's wallet.
# SDK Methods Guide
Detailed SDK interface handbook for interacting with the Somnia Data Streams protocol via the typescript SDK
Somnia Data Streams is the on-chain data streaming protocol that powers real-time, composable applications on the Somnia Network. It is available as an SDK Package [@somnia-chain/streams](https://www.npmjs.com/package/@somnia-chain/streams).
This SDK exposes all the core functionality developers need to write, read, subscribe to, and manage Data Streams and events directly from their dApps.
Before using the Data Streams SDK, ensure you have a working Node.js or Next.js environment (Node 18+ recommended). You’ll need access to a Somnia RPC endpoint (Testnet or Mainnet) and a wallet private key for publishing data.
### Installation
```bash
npm i @somnia-chain/streams viem dotenv
```
The SDK depends on [viem](https://viem.sh/) for blockchain interactions.
### Project Setup
Create a `.env.local` or `.env` file in your project root:
```bash
RPC_URL=https://dream-rpc.somnia.network
PRIVATE_KEY=your_private_key_here
```
⚠️ Never expose private keys in client-side code. Keep writes (publishing data) in server routes or backend environments.
### Basic Initialization
You’ll typically use two clients:
* A public client for reading and subscribing
* A wallet client for writing to the Somnia chain
```typescript
import { SDK } from '@somnia-chain/streams'
import { createPublicClient, createWalletClient, http } from 'viem'
import { privateKeyToAccount } from 'viem/accounts'
import { somniaTestnet } from 'viem/chains'
const rpcUrl = process.env.RPC_URL!
const account = privateKeyToAccount(process.env.PRIVATE_KEY as `0x${string}`)
const sdk = new SDK({
public: createPublicClient({ chain: somniaTestnet, transport: http(rpcUrl) }),
wallet: createWalletClient({ chain: somniaTestnet, account, transport: http(rpcUrl) })
})
```
### Somnia Data Streams
Data Streams in Somnia represent structured, verifiable data channels. Every piece of data conforms to a schema that defines its structure (e.g. `timestamp`, `content`, `sender`), and publishers can emit this data either as on-chain transactions or off-chain event notifications.
The SDK follows a simple pattern:
```typescript
const sdk = new SDK({
public: getPublicClient(), // for reading and subscriptions
wallet: getWalletClient() // for writing
})
```
You’ll interact primarily through the `sdk.streams` interface.
### Core Methods Overview
### Write
#### `set(d: DataStream[]): Promise<Hex | null>`
**Description**
Publishes one or more data streams to the Somnia blockchain. Each stream must specify a `schema ID`, a `unique ID`, and the `encoded payload`.
**Use Case**
When you want to store data on-chain in a standardized format (e.g., chat messages, sensor telemetry, or leaderboard updates).
**Example**
```typescript
const tx = await sdk.streams.set([
{ id: dataId, schemaId, data }
])
console.log('Data published with tx hash:', tx)
```
Always register your schema before calling `set()` ; otherwise, the transaction will revert.
#### `emitEvents(e: EventStream[]): Promise<Hex | Error | null>`
**Description**
Emits a registered Streams event without persisting new data. This is used for off-chain reactivity, triggering listeners subscribed via WebSocket.
**Example**
```typescript
await sdk.streams.emitEvents([
{
id: 'ChatMessage',
argumentTopics: [topic],
data: '0x' // optional encoded payload
}
])
```
Common Use includes notifying subscribers when something happens, e.g., “new message sent” or “order filled”.
#### `setAndEmitEvents(d: DataStream[], e: EventStream[]): Promise<Hex | Error | null>`
**Description**
Performs an atomic on-chain operation that both writes data and emits a corresponding event. This ensures your data and notifications are always in sync.
**Example**
```typescript
await sdk.streams.setAndEmitEvents(
[{ id: dataId, schemaId, data }],
[{ id: 'ChatMessage', argumentTopics: [topic], data: '0x' }]
)
```
It is ideal for chat apps, game updates, or IoT streams — where data must be recorded and instantly broadcast.
### Manage
#### `registerDataSchemas(registrations: DataSchemaRegistration[], ignoreRegisteredSchemas?: boolean): Promise<Hex | Error | null>`
**Description**
Registers a new data schema on-chain. Schemas define the structure of your Streams data, like a table schema in a database. The optional `ignoreRegisteredSchemas` parameter allows skipping registration if the schema is already registered.
**Example**
```typescript
await sdk.streams.registerDataSchemas([
{
schemaName: "chat",
schema: 'uint64 timestamp, string message, address sender',
parentSchemaId: zeroBytes32 // root schema
}
], true) // Optionally ignore if already registered
```
Register before writing any new data type. If you modify the schema structure later, register it again as a new schema version.
#### `registerEventSchemas(ids: string[], schemas: EventSchema[]): Promise<Hex | Error | null>`
**Description**
Registers event definitions that can later be emitted or subscribed to.
**Example**
```typescript
await sdk.streams.registerEventSchemas(
['ChatMessage'],
[{
params: [{ name: 'roomId', paramType: 'bytes32', isIndexed: true }],
eventTopic: 'ChatMessage(bytes32 indexed roomId)'
}]
)
```
Use before calling `emitEvents()` or `subscribe()` for a specific event.
#### `manageEventEmittersForRegisteredStreamsEvent(streamsEventId: string, emitter: Address, isEmitter: boolean): Promise<Hex | Error | null>`
**Description**
Grants or revokes permission for an address to emit a specific event.
**Example**
```typescript
await sdk.streams.manageEventEmittersForRegisteredStreamsEvent(
'ChatMessage',
'0x1234abcd...',
true // allow this address to emit
)
```
Used for access control in multi-publisher systems.
### Read
#### `getByKey(schemaId: SchemaID, publisher: Address, key: Hex): Promise<Hex[] | SchemaDecodedItem[][] | null>`
**Description**
Retrieves data stored under a schema by its unique ID.
**Example**
```typescript
const msg = await sdk.streams.getByKey(schemaId, publisher, dataId)
console.log('Data:', msg)
```
An example includes fetching a specific record, e.g., “fetch message by message ID”.
#### `getAtIndex(schemaId: SchemaID, publisher: Address, idx: bigint): Promise<Hex[] | SchemaDecodedItem[][] | null>`
**Description**
Fetches the record at a given index (0-based).
**Example**
```typescript
const record = await sdk.streams.getAtIndex(schemaId, publisher, 0n)
```
It is useful for sequential datasets like logs or telemetry streams.
#### `getBetweenRange(schemaId: SchemaID, publisher: Address, startIndex: bigint, endIndex: bigint): Promise<Hex[] | SchemaDecodedItem[][] | Error | null>`
**Description**
Fetches records within a specified index range (0-based, inclusive start, exclusive end).
**Use Case**
Retrieving a batch of historical data, such as paginated logs or time-series entries.
**Example**
```typescript
const records = await sdk.streams.getBetweenRange(schemaId, publisher, 0n, 10n)
console.log('Records in range:', records)
```
#### `getAllPublisherDataForSchema(schemaReference: SchemaReference, publisher: Address): Promise<Hex[] | SchemaDecodedItem[][] | null>`
**Description**
Retrieves all data published by a specific address under a given schema.
**Use Case**
Fetching complete datasets for analysis or synchronization.
**Example**
```typescript
const allData = await sdk.streams.getAllPublisherDataForSchema(schemaReference, publisher)
console.log('All publisher data:', allData)
```
#### `getLastPublishedDataForSchema(schemaId: SchemaID, publisher: Address): Promise<Hex[] | SchemaDecodedItem[][] | null>`
**Description**
Retrieves the most recently published data under a schema by a publisher.
**Use Case**
Getting the latest update, such as the most recent sensor reading or message.
**Example**
```typescript
const latest = await sdk.streams.getLastPublishedDataForSchema(schemaId, publisher)
console.log('Latest data:', latest)
```
#### `totalPublisherDataForSchema(schemaId: SchemaID, publisher: Address): Promise<bigint | null>`
**Description**
Returns how many records a publisher has stored under a schema.
**Example**
```typescript
const total = await sdk.streams.totalPublisherDataForSchema(schemaId, publisher)
console.log(`Total entries: ${total}`)
```
#### `isDataSchemaRegistered(schemaId: SchemaID): Promise<boolean | null>`
**Description**
Checks if a schema exists on-chain.
**Example**
```typescript
const exists = await sdk.streams.isDataSchemaRegistered(schemaId)
if (!exists) console.log('Schema not found')
```
#### `parentSchemaId(schemaId: SchemaID): Promise<Hex | null>`
**Description**
Finds the parent schema of a given schema, if one exists.
**Example**
```typescript
const parent = await sdk.streams.parentSchemaId(schemaId)
console.log('Parent Schema ID:', parent)
```
#### `schemaIdToId(schemaId: SchemaID): Promise<string | null>`
**Description**
Converts a schema ID (Hex) to its corresponding string identifier.
**Use Case**
Mapping hashed IDs back to human-readable names for display or logging.
**Example**
```typescript
const id = await sdk.streams.schemaIdToId(schemaId)
console.log('Schema ID string:', id)
```
#### `idToSchemaId(id: string): Promise<Hex | null>`
**Description**
Converts a string identifier to its corresponding schema ID (Hex).
**Use Case**
Looking up hashed IDs from known names for queries.
**Example**
```typescript
const schemaId = await sdk.streams.idToSchemaId('chat')
console.log('Schema ID:', schemaId)
```
#### `getAllSchemas(): Promise<string[] | null>`
**Description**
Retrieves a list of all registered schema identifiers.
**Use Case**
Discovering available schemas in the protocol.
**Example**
```typescript
const schemas = await sdk.streams.getAllSchemas()
console.log('All schemas:', schemas)
```
#### `getEventSchemasById(ids: string[]): Promise<EventSchema[] | null>`
**Description**
Fetches event schema details for given identifiers.
**Use Case**
Inspecting registered event structures before subscribing or emitting.
**Example**
```typescript
const eventSchemas = await sdk.streams.getEventSchemasById(['ChatMessage'])
console.log('Event schemas:', eventSchemas)
```
#### `computeSchemaId(schema: string): Promise<Hex | null>`
**Description**
Computes the deterministic `schemaId` without registering it.
**Example**
```typescript
const schemaId = await sdk.streams.computeSchemaId('uint64 timestamp, string content')
```
#### `getSchemaFromSchemaId(schemaId: SchemaID): Promise<{ baseSchema: string, finalSchema: string, schemaId: Hex } | Error | null>`
**Description**
Request a schema given the schema id used for data publishing and let the SDK take care of schema extensions.
**Use Case**
Retrieving schema details, including the base schema and the final extended schema, for a given schema ID.
**Example**
```typescript
const schemaInfo = await sdk.streams.getSchemaFromSchemaId(schemaId)
console.log('Schema info:', schemaInfo)
```
### Helpers
#### `deserialiseRawData(rawData: Hex[], parentSchemaId: Hex, schemaLookup: { schema: string; schemaId: Hex; } | null): Promise<Hex[] | SchemaDecodedItem[][] | null>`
**Description**
Deserializes raw data using the provided schema information.
**Use Case**
Decoding fetched raw bytes into structured objects for application use.
**Example**
```typescript
const decoded = await sdk.streams.deserialiseRawData(rawData, parentSchemaId, schemaLookup)
console.log('Decoded data:', decoded)
```
### Subscribe
#### `subscribe(initParams: SubscriptionInitParams): Promise<{ subscriptionId: string, unsubscribe: () => void } | undefined>`
**Description**
Creates a real-time WebSocket subscription to a Streams event. Whenever the specified event fires, the SDK calls your `onData` callback — optionally including enriched data from on-chain calls.
**Parameters**
* `ethCalls`: Fixed set of ETH calls that must be executed before onData callback is triggered. Multicall3 is recommended. Can be an empty array.
* `context`: Event sourced selectors to be added to the data field of ETH calls, possible values: topic0, topic1, topic2, topic3, topic4, data and address.
* `onData`: Callback for a successful reactivity notification.
* `onError`: Callback for a failed attempt.
* `eventContractSource`: Optional but is the contract event source (any on Somnia) that will be emitting the logs specified by topicOverrides.
* `topicOverrides`: Optional but this argument is a filter applied to the subscription. Up to 4 bytes32 event topics can be supplied. By not defining, this is the equivalent of a wildcard subscription to all event topics
* `onlyPushChanges`: Whether the data should be pushed to the subscriber only if eth\_call results are different from the previous.
**Example**
```typescript
// Wildcard subscription to all events emitted by all contracts
await sdk.streams.subscribe({
ethCalls: [], // No view calls
onData: (data) => {}
})
```
**With `ethCalls`**
```typescript
import { toEventSelector } from "viem"
const transferSelector = toEventSelector({
name: 'Transfer',
type: 'event',
inputs: [
{ type: 'address', indexed: true, name: 'from' },
{ type: 'address', indexed: true, name: 'to' },
{ type: 'uint256', indexed: false, name: 'value' }
]
})
await sdk.streams.subscribe({
topicOverrides: [
transferSelector, // Topic 0 (Transfer event)
],
ethCalls: [{
to: '0xERC20Address',
data: encodeFunctionData({
abi: erc20Abi,
functionName: 'balanceOf',
args: ['0xUserAddress']
})
}],
onData: (data) => console.log('Trade + balance data:', data)
})
```
Useful for off-chain reactivity: real-time dashboards, chat updates, live feeds, or notifications.
**Notes**
* Requires `createPublicClient({ transport: webSocket() })`
* Use `setAndEmitEvents()` on the publisher side to trigger matching subscriptions.
### Protocol
#### `getSomniaDataStreamsProtocolInfo(): Promise<GetSomniaDataStreamsProtocolInfoResponse | Error | null>`
**Description**
Retrieves information about the Somnia Data Streams protocol.
**Use Case**
Fetching protocol-level details, such as version or configurati프로젝트로 돌아가기
