bitpro가 원문에서 글자만 모은 내용이에요. 표·그림·수식은 빠지거나 깨질 수 있어요. 전체 54쪽 중 15쪽
: somniaTestnet,
transport: http(getEnv('RPC_URL_SOMNIA')),
})
export const somniaSdk = new SDK({
public: somniaPublicClient,
wallet: somniaWalletClient,
})
// === Client 2: Sepolia Public Client (Read-Only) ===
export const sepoliaPublicClient = createPublicClient({
chain: sepolia,
transport: http(getEnv('RPC_URL_SEPOLIA')),
})
```
## Define the Price Feed Schema
Our schema will store the core data from Chainlink's feed.
**`src/lib/schema.ts`**
```typescript
// This schema will store historical price snapshots
export const priceFeedSchema =
'uint64 timestamp, int256 price, uint80 roundId, string pair'
```
* `timestamp`: The `updatedAt` time from Chainlink.
* `price`: The `answer` (e.g., ETH price).
* `roundId`: The Chainlink round ID, to prevent duplicates.
* `pair`: A string to identify the feed (e.g., "ETH/USD").
## Create the Chainlink Reader
Let's create a dedicated file to handle fetching data from Chainlink. We will use the `ETH/USD` feed on Sepolia.
**`src/lib/chainlinkReader.ts`**
```typescript
import { parseAbi, Address } from 'viem'
import { sepoliaPublicClient } from './clients'
// Chainlink ETH/USD Feed on Sepolia Testnet
const CHAINLINK_FEED_ADDRESS: Address = '0x694AA1769357215DE4FAC081bf1f309aDC325306'
// Minimal ABI for AggregatorV3Interface
const CHAINLINK_ABI = parseAbi([
'function latestRoundData() external view returns (uint80 roundId, int256 answer, uint256 startedAt, uint256 updatedAt, uint80 answeredInRound)',
'function decimals() external view returns (uint8)',
])
export interface PriceData {
roundId: bigint
price: bigint
timestamp: bigint
decimals: number
}
/**
* Fetches the latest price data from the Chainlink ETH/USD feed on Sepolia.
*/
export async function fetchLatestPrice(): Promise<PriceData> {
console.log('Fetching latest price from Chainlink on Sepolia...')
try {
const [roundData, decimals] = await Promise.all([
sepoliaPublicClient.readContract({
address: CHAINLINK_FEED_ADDRESS,
abi: CHAINLINK_ABI,
functionName: 'latestRoundData',
}),
sepoliaPublicClient.readContract({
address: CHAINLINK_FEED_ADDRESS,
abi: CHAINLINK_ABI,
functionName: 'decimals',
})
])
const [roundId, answer, , updatedAt] = roundData
console.log(`Chainlink data received: Round ${roundId}, Price ${answer}`)
return {
roundId,
price: answer,
timestamp: updatedAt,
decimals,
}
} catch (error: any) {
console.error(`Failed to read from Chainlink: ${error.message}`)
throw error
}
}
```
## Build the Snapshot Bot (The Hybrid App)
This is the core of our project. This script will:
1. Fetch the latest price from Chainlink (using our module).
2. Encode this data using our `priceFeedSchema`.
3. Publish the data to Somnia Data Streams.
**`src/scripts/snapshotBot.ts`**
<pre class="language-typescript"><code class="lang-typescript">import 'dotenv/config'
import { somniaSdk } from '../lib/clients'
import { priceFeedSchema } from '../lib/schema'
import { fetchLatestPrice } from '../lib/chainlinkReader'
import { SchemaEncoder, zeroBytes32 } from '@somnia-chain/streams'
import { toHex, Hex } from 'viem'
import { waitForTransactionReceipt } from 'viem/actions'
const PAIR_NAME = "ETH/USD"
async function main() {
console.log('--- Starting Snapshot Bot ---')
// 1. Initialize SDK and Encoder
const sdk = somniaSdk
const encoder = new SchemaEncoder(priceFeedSchema)
const publisherAddress = sdk.wallet.account?.address
if (!publisherAddress) throw new Error('Wallet client not initialized.')
// 2. Compute Schema ID and Register (idempotent)
const schemaId = await sdk.streams.computeSchemaId(priceFeedSchema)
if (!schemaId) throw new Error('Could not compute schemaId')
const ignoreAlreadyRegisteredSchemas = true
<strong> const regTx = await sdk.streams.registerDataSchemas([
</strong> { id: 'price-feed-v1', schema: priceFeedSchema }
], ignoreAlreadyRegisteredSchemas)
if (!regTx) throw new Error('Failed to register schema')
await waitForTransactionReceipt(sdk.public, { hash: regTx })
// 3. Fetch data from Chainlink
const priceData = await fetchLatestPrice()
// 4. Encode data for Somnia Streams
const encodedData: Hex = encoder.encodeData([
{ name: 'timestamp', value: priceData.timestamp.toString(), type: 'uint64' },
{ name: 'price', value: priceData.price.toString(), type: 'int256' },
{ name: 'roundId', value: priceData.roundId.toString(), type: 'uint80' },
{ name: 'pair', value: PAIR_NAME, type: 'string' },
])
// 5. Create a unique Data ID (using the roundId to prevent duplicates)
const dataId = toHex(`price-${PAIR_NAME}-${priceData.roundId}`, { size: 32 })
// 6. Publish to Somnia Data Streams
console.log(`Publishing price data to Somnia Streams...`)
const txHash = await sdk.streams.set([
{ id: dataId, schemaId, data: encodedData }
])
if (!txHash) throw new Error('Failed to publish to Streams')
await waitForTransactionReceipt(sdk.public, { hash: txHash })
console.log('\n--- Snapshot Complete! ---')
console.log(` Publisher: ${publisherAddress}`)
console.log(` Schema ID: ${schemaId}`)
console.log(` Data ID: ${dataId}`)
console.log(` Tx Hash: ${txHash}`)
}
main().catch((e) => {
console.error(e)
process.exit(1)
})
</code></pre>
**To run your bot:**
Add a script to `package.json`: `"snapshot": "ts-node src/scripts/snapshotBot.ts"`
Run it: `npm run snapshot`
You can run this script multiple times. It will only add new data if Chainlink's `roundId` has changed.
## Read Your Historical Price Feed
Now for the payoff. Let's create a script that reads our new on-chain history from Somnia Streams.
**`src/scripts/readHistory.ts`**
```typescript
import 'dotenv/config'
import { somniaSdk } from '../lib/clients'
import { priceFeedSchema } from '../lib/schema'
import { SchemaDecodedItem } from '@somnia-chain/streams'
// Helper to decode the SDK's output
interface PriceRecord {
timestamp: number
price: bigint
roundId: bigint
pair: string
}
function decodePriceRecord(row: SchemaDecodedItem[]): PriceRecord {
const val = (field: any) => field?.value?.value ?? field?.value ?? ''
return {
timestamp: Number(val(row[0])),
price: BigInt(val(row[1])),
roundId: BigInt(val(row[2])),
pair: String(val(row[3])),
}
}
async function main() {
console.log('--- Reading Historical Price Feed from Somnia Streams ---')
const sdk = somniaSdk
// Use the *publisher address* from your .env file
const publisherAddress = sdk.wallet.account?.address
if (!publisherAddress) throw new Error('Wallet client not initialized.')
const schemaId = await sdk.streams.computeSchemaId(priceFeedSchema)
if (!schemaId) throw new Error('Could not compute schemaId')
console.log(`Reading all data for publisher: ${publisherAddress}`)
console.log(`Schema: ${schemaId}\n`)
// Fetch all data for this schema and publisher
const data = await sdk.streams.getAllPublisherDataForSchema(schemaId, publisherAddress)
if (!data || data.length === 0) {
console.log('No price history found. Run the snapshot bot first.')
return
}
const records = (data as SchemaDecodedItem[][]).map(decodePriceRecord)
// Sort by timestamp
records.sort((a, b) => a.timestamp - b.timestamp)
console.log(`Found ${records.length} historical price points:\n`)
records.forEach(record => {
// We assume the decimals are 8 for this ETH/USD feed
const priceFloat = Number(record.price) / 10**8
console.log(
`[${new Date(record.timestamp * 1000).toISOString()}] ${record.pair} - $${priceFloat.toFixed(2)} (Round: ${record.roundId})`
)
})
}
main().catch((e) => {
console.error(e)
process.exit(1)
})
```
**To read the history:**
Add to `package.json`: `"history": "ts-node src/scripts/readHistory.ts"`
Run it: `npm run history`
**Expected Output:**
```bash
--- Reading Historical Price Feed from Somnia Streams ---
...
Found 3 historical price points:
[2025-11-06T14:30:00.000Z] ETH/USD - $3344.50 (Round: 110...)
[2025-11-06T14:35:00.000Z] ETH/USD - $3362.12 (Round: 111...)
[2025-11-06T14:40:00.000Z] ETH/USD - $3343.90 (Round: 112...)
```
## Conclusion: Key Takeaways
You have successfully built a hybrid, cross-chain application.
* You combined an **external "truth source"** (Chainlink) with Somnia's **composable storage layer** (Somnia Data Streams).
* You created a new, valuable, on-chain data product: a **historical, queryable price feed** that any dApp on Somnia can now read from and trust.
* You demonstrated the power of the `publisher` address as a verifiable source. Any dApp can now consume your feed, knowing it was published by *your* trusted bot.
This pattern can be extended to any external data source: weather, sports results, IoT data, and more. You can run the `snapshotBot.ts` script as a cron job or serverless function to create a truly autonomous, on-chain oracle.
# Working with Multiple Publishers in a Shared Stream
The core architecture of Somnia Data Streams decouples data schemas from publishers. This allows multiple different accounts (or devices) to publish data using the **same schema**. The data conforms to the same data structure, regardless of who published it.
This model is perfect for multi-source data scenarios, such as:
* **Multi-User Chat:** Multiple users sending messages under the same `chatMessage` schema.
* **IoT (Internet of Things):** Hundreds of sensors submitting data under the same `telemetry` schema.
* **Gaming:** All players in a game publishing their positions and scores under the same `playerUpdate` schema.
In this tutorial, we will demonstrate how to build an "aggregator" application that collects and merges data from two different "devices" (two separate wallet accounts) publishing to the same "telemetry" schema.
## Prerequisites
* Node.js 20+
* `@somnia-chain/streams` and `viem` libraries installed
* An `RPC_URL` for access to the Somnia Testnet
* **Two (2)** funded Somnia Testnet wallets for publishing data.
## What You’ll Build
In this tutorial, we will build two main components:
1. A **Publisher Script** that simulates two different wallets sending data to the same telemetry schema.
2. An **Aggregator Script** that fetches *all* data from a specified list of publishers, merges them into a single list, and sorts them by timestamp.
## Project Setup
Create a new directory for your application and install the necessary packages.
```bash
mkdir somnia-aggregator
cd somnia-aggregator
npm init -y
npm i @somnia-chain/streams viem dotenv
npm i -D @types/node typescript ts-node
```
Create a `tsconfig.json` file in your project root:
```json
{
"compilerOptions": {
"target": "ES2020",
"module": "commonjs",
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"outDir": "./dist"
},
"include": ["src/**/*"]
}
```
## Configure Environment Variables
Create a `.env` file in your project root. For this tutorial, we will need **two** different private keys.
```bash
# .env
RPC_URL=https://dream-rpc.somnia.network/
# Simulates two different devices/publishers
PUBLISHER_1_PK=0xPUBLISHER_ONE_PRIVATE_KEY
PUBLISHER_2_PK=0xPUBLISHER_TWO_PRIVATE_KEY
```
{% hint style="info" %}
**IMPORTANT:** Never expose private keys in client-side (browser) code or public repositories. The scripts in this tutorial are intended to be run server-side.
{% endhint %}
## Chain and Client Configuration
Create a folder named `src/lib` and set up your `chain.ts` and `clients.ts` files.
**`src/lib/chain.ts`**
```typescript
import { defineChain } from 'viem'
export const somniaTestnet = defineChain({
id: 50312,
name: 'Somnia Testnet',
network: 'somnia-testnet',
nativeCurrency: { name: 'STT', symbol: 'STT', decimals: 18 },
rpcUrls: {
default: { http: ['[https://dream-rpc.somnia.network]'] },
public: { http: ['[https://dream-rpc.somnia.network]'] },
},
} as const)
```
**`src/lib/clients.ts`**
```typescript
import 'dotenv/config'
import { createPublicClient, createWalletClient, http, PublicClient } from 'viem'
import { privateKeyToAccount, PrivateKeyAccount } from 'viem/accounts'
import { somniaTestnet } from './chain'
function getEnv(key: string): string {
const value = process.env[key]
if (!value) {
throw new Error(`Missing environment variable: ${key}`)
}
return value
}
// A single Public Client for read operations
export const publicClient: PublicClient = createPublicClient({
chain: somniaTestnet,
transport: http(getEnv('RPC_URL')),
})
// Two different Wallet Clients for simulation
export const walletClient1 = createWalletClient({
account: privateKeyToAccount(getEnv('PUBLISHER_1_PK') as `0x${string}`),
chain: somniaTestnet,
transport: http(getEnv('RPC_URL')),
})
export const walletClient2 = createWalletClient({
account: privateKeyToAccount(getEnv('PUBLISHER_2_PK') as `0x${string}`),
chain: somniaTestnet,
transport: http(getEnv('RPC_URL')),
})
```
## Define the Shared Schema
Let's define the common schema that all publishers will use.
**`src/lib/schema.ts`**
```typescript
// This schema will be used by multiple devices
export const telemetrySchema =
'uint64 timestamp, string deviceId, int32 x, int32 y, uint32 speed'
```
## Create the Publisher Script
Now, let's create a script that simulates how two different publishers will send data to this schema. This script will take which publisher to use as a command-line argument.
**`src/scripts/publishData.ts`**
```typescript
import 'dotenv/config'
import { SDK, SchemaEncoder, zeroBytes32 } from '@somnia-chain/streams'
import { publicClient, walletClient1, walletClient2 } from '../lib/clients'
import { telemetrySchema } from '../lib/schema'
import { toHex, Hex, WalletClient } from 'viem'
import { waitForTransactionReceipt } from 'viem/actions'
// Select which publisher to use
async function getPublisher(): Promise<{ client: WalletClient, deviceId: string }> {
const arg = process.argv[2] // 'p1' or 'p2'
if (arg === 'p2') {
console.log('Using Publisher 2 (Device B)')
return { client: walletClient2, deviceId: 'device-b-002' }
}
console.log('Using Publisher 1 (Device A)')
return { client: walletClient1, deviceId: 'device-a-001' }
}
// Helper function to encode the data
function encodeTelemetry(encoder: SchemaEncoder, deviceId: string): Hex {
const now = Date.now().toString()
return encoder.encodeData([
{ name: "timestamp", value: now, type: "uint64" },
{ name: "deviceId", value: deviceId, type: "string" },
{ name: "x", value: Math.floor(Math.random() * 1000).toString(), type: "int32" },
{ name: "y", value: Math.floor(Math.random() * 1000).toString(), type: "int32" },
{ name: "speed", value: Math.floor(Math.random() * 120).toString(), type: "uint32" },
])
}
async function main() {
const { client, deviceId } = await getPublisher()
const publisherAddress = client.account.address
console.log(`Publisher Address: ${publisherAddress}`)
const sdk = new SDK({ public: publicClient, wallet: client })
const encoder = new SchemaEncoder(telemetrySchema)
// 1. Compute the Schema ID
const schemaId = await sdk.streams.computeSchemaId(telemetrySchema)
if (!schemaId) throw new Error('Could not compute schemaId')
console.log(`Schema ID: ${schemaId}`)
// 2. Register Schema (Updated for new API)
console.log('Registering schema (if not already registered)...')
const ignoreAlreadyRegisteredSchemas = true
const regTx = await sdk.streams.registerDataSchemas([
{
schemaName: 'telemetry', // Updated: 'id' is now 'schemaName'
schema: telemetrySchema,
parentSchemaId: zeroBytes32
}
], ignoreAlreadyRegisteredSchemas)
if (regTx) {
console.log('Schema registration transaction sent:', regTx)
await waitForTransactionReceipt(publicClient, { hash: regTx })
console.log('Schema registered successfully!')
} else {
console.log('Schema was already registered. No transaction sent.')
}
// 3. Encode the data
const encodedData = encodeTelemetry(encoder, deviceId)
// 4. Publish the data
// We make the dataId unique with a timestamp and device ID
const dataId = toHex(`${deviceId}-${Date.now()}`, { size: 32 })
const txHash = await sdk.streams.set([
{ id: dataId, schemaId, data: encodedData }
])
if (!txHash) throw new Error('Failed to publish data')
console.log(`Publishing data... Tx: ${txHash}`)
await waitForTransactionReceipt(publicClient, { hash: txHash })
console.log('Data published successfully!')
}
main().catch((e) => {
console.error(e)
process.exit(1)
})
```
**To run this script:**
Add the following scripts to your `package.json` file:
```json
"scripts": {
"publish:p1": "ts-node src/scripts/publishData.ts p1",
"publish:p2": "ts-node src/scripts/publishData.ts p2"
}
```
Now, open two different terminals and send data from each:
```bash
# Terminal 1
npm run publish:p1
# Terminal 2
npm run publish:p2
```
Repeat this a few times to build up a dataset from both publishers.
## Create the Aggregator Script
Now that we have published our data, we can write the "aggregator" script that collects, merges, and sorts all data from these two (or more) publishers.
**`src/scripts/aggregateData.ts`**
```typescript
import 'dotenv/config'
import { SDK, SchemaDecodedItem } from '@somnia-chain/streams'
import { publicClient, walletClient1, walletClient2 } from '../lib/clients'
import { telemetrySchema } from '../lib/schema'
import { Address } from 'viem'
// LIST OF PUBLISHERS TO TRACK
// You could also fetch this list dynamically (e.g., from a contract or database).
const TRACKED_PUBLISHERS: Address[] = [
walletClient1.account.address,
walletClient2.account.address,
]
// Helper function to convert SDK data into a cleaner object
// (Similar to the 'val' function in the Minimal On-Chain Chat App Tutorial)
function decodeTelemetryRecord(row: SchemaDecodedItem[]): TelemetryRecord {
const val = (field: any) => field?.value?.value ?? field?.value ?? ''
return {
timestamp: Number(val(row[0])),
deviceId: String(val(row[1])),
x: Number(val(row[2])),
y: Number(val(row[3])),
speed: Number(val(row[4])),
}
}
// Type definition for our data
interface TelemetryRecord {
timestamp: number
deviceId: string
x: number
y: number
speed: number
publisher?: Address // We will add this field later
}
async function main() {
// The aggregator doesn't need to write data, so it only uses the publicClient
const sdk = new SDK({ public: publicClient })
const schemaId = await sdk.streams.computeSchemaId(telemetrySchema)
if (!schemaId) throw new Error('Could not compute schemaId')
console.log(`Aggregator started. Tracking ${TRACKED_PUBLISHERS.length} publishers...`)
console.log(`Schema ID: ${schemaId}\n`)
const allRecords: TelemetryRecord[] = []
// 1. Loop through each publisher
for (const publisherAddress of TRACKED_PUBLISHERS) {
console.log(`--- Fetching data for ${publisherAddress} ---`)
// 2. Fetch all data for the publisher based on the schema
// Note: The SDK automatically decodes the data if the schema is registered
const data = await sdk.streams.getAllPublisherDataForSchema(schemaId, publisherAddress)
if (!data || data.length === 0) {
console.log('No data found for this publisher.\n')
continue
}
// 3. Transform the data and add the 'publisher' field
const records: TelemetryRecord[] = (data as SchemaDecodedItem[][]).map(row => ({
...decodeTelemetryRecord(row),
publisher: publisherAddress // To know where the data came from
}))
console.log(`Found ${records.length} records.`)
// 4. Add all records to the main list
allRecords.push(...records)
}
// 5. Sort all data by timestamp
console.log('\n--- Aggregation Complete ---')
console.log(`Total records fetched: ${allRecords.length}`)
allRecords.sort((a, b) => a.timestamp - b.timestamp)
// 6. Display the result
console.log('\n--- Combined and Sorted Telemetry Log ---')
allRecords.forEach(record => {
console.log(
`[${new Date(record.timestamp).toISOString()}] [${record.publisher}] - Device: ${record.deviceId}, Speed: ${record.speed}`
)
})
}
main().catch((e) => {
console.error(e)
process.exit(1)
})
```
**To run this script:**
Add the scr프로젝트로 돌아가기
