DeBros/network
4
0
Fork 0
mirror of https://github.com/DeBrosOfficial/network.git synced 2025-10-06 12:09:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Removed wrong kv store from network

Browse Source
This commit is contained in:
anonpenguin 2025-09-12 08:02:45 +03:00
parent b53ea93389
commit f580e3d8f4
14 changed files with 6957 additions and 754 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
.zed/debug.json Normal file
View File

@ -0,0 +1,31 @@
// Project-local debug tasks
//
// For more documentation on how to configure debug tasks,
// see: https://zed.dev/docs/debugger
[
{
"label": "Gateway Go (Delve)",
"adapter": "Delve",
"request": "launch",
"mode": "debug",
"program": "./cmd/gateway",
"env": {
"GATEWAY_ADDR": ":8080",
"GATEWAY_BOOTSTRAP_PEERS": "/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWSHHwEY6cga3ng7tD1rzStAU58ogQXVMX3LZJ6Gqf6dee",
"GATEWAY_NAMESPACE": "default",
"GATEWAY_API_KEY": "ak_iGustrsFk9H8uXpwczCATe5U:default"
}
},
{
"label": "E2E Test Go (Delve)",
"adapter": "Delve",
"request": "launch",
"mode": "test",
"buildFlags": "-tags e2e",
"program": "./e2e",
"env": {
"GATEWAY_API_KEY": "ak_iGustrsFk9H8uXpwczCATe5U:default"
},
"args": ["-test.v"]
}
]

88
cmd/cli/main.go
View File

@ -72,8 +72,6 @@ func main() {
os.Exit(1) os.Exit(1)
} }
handleQuery(args[0]) handleQuery(args[0])
case "storage":
handleStorage(args)
case "pubsub": case "pubsub":
handlePubSub(args) handlePubSub(args)
case "connect": case "connect":
@ -215,78 +213,6 @@ func handleQuery(sql string) {
} }
} }
func handleStorage(args []string) {
if len(args) == 0 {
fmt.Fprintf(os.Stderr, "Usage: network-cli storage <get|put|list> [args...]\n")
os.Exit(1)
}
// Ensure user is authenticated
_ = ensureAuthenticated()
client, err := createClient()
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to create client: %v\n", err)
os.Exit(1)
}
defer client.Disconnect()
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
subcommand := args[0]
switch subcommand {
case "get":
if len(args) < 2 {
fmt.Fprintf(os.Stderr, "Usage: network-cli storage get <key>\n")
os.Exit(1)
}
value, err := client.Storage().Get(ctx, args[1])
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to get value: %v\n", err)
os.Exit(1)
}
// Try to decode if it looks like base64
decoded := tryDecodeBase64(string(value))
fmt.Printf("%s\n", decoded)
case "put":
if len(args) < 3 {
fmt.Fprintf(os.Stderr, "Usage: network-cli storage put <key> <value>\n")
os.Exit(1)
}
err := client.Storage().Put(ctx, args[1], []byte(args[2]))
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to store value: %v\n", err)
os.Exit(1)
}
fmt.Printf("✅ Stored key: %s\n", args[1])
case "list":
prefix := ""
if len(args) > 1 {
prefix = args[1]
}
keys, err := client.Storage().List(ctx, prefix, 100)
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to list keys: %v\n", err)
os.Exit(1)
}
if format == "json" {
printJSON(keys)
} else {
for _, key := range keys {
fmt.Println(key)
}
}
default:
fmt.Fprintf(os.Stderr, "Unknown storage command: %s\n", subcommand)
os.Exit(1)
}
}
func handlePubSub(args []string) { func handlePubSub(args []string) {
if len(args) == 0 { if len(args) == 0 {
fmt.Fprintf(os.Stderr, "Usage: network-cli pubsub <publish|subscribe|topics> [args...]\n") fmt.Fprintf(os.Stderr, "Usage: network-cli pubsub <publish|subscribe|topics> [args...]\n")
@ -371,8 +297,6 @@ func handlePubSub(args []string) {
} }
} }
// ensureAuthenticated ensures the user has valid credentials for the gateway
// Returns the credentials or exits the program on failure
func ensureAuthenticated() *auth.Credentials { func ensureAuthenticated() *auth.Credentials {
gatewayURL := auth.GetDefaultGatewayURL() gatewayURL := auth.GetDefaultGatewayURL()
@ -401,7 +325,6 @@ func openBrowser(target string) error {
return nil return nil
} }
// getenvDefault returns env var or default if empty/undefined.
func getenvDefault(key, def string) string { func getenvDefault(key, def string) string {
if v := strings.TrimSpace(os.Getenv(key)); v != "" { if v := strings.TrimSpace(os.Getenv(key)); v != "" {
return v return v
@ -543,7 +466,6 @@ func createClient() (client.NetworkClient, error) {
return networkClient, nil return networkClient, nil
} }
// discoverBootstrapPeer tries to find the bootstrap peer from saved peer info
func discoverBootstrapPeer() string { func discoverBootstrapPeer() string {
// Look for peer info in common locations // Look for peer info in common locations
peerInfoPaths := []string{ peerInfoPaths := []string{
@ -568,7 +490,6 @@ func discoverBootstrapPeer() string {
return "" // Return empty string if no peer info found return "" // Return empty string if no peer info found
} }
// tryDecodeBase64 attempts to decode a string as base64, returns original if not valid base64
func tryDecodeBase64(s string) string { func tryDecodeBase64(s string) string {
// Only try to decode if it looks like base64 (no spaces, reasonable length) // Only try to decode if it looks like base64 (no spaces, reasonable length)
if len(s) > 0 && len(s)%4 == 0 && !strings.ContainsAny(s, " \n\r\t") { if len(s) > 0 && len(s)%4 == 0 && !strings.ContainsAny(s, " \n\r\t") {
@ -583,7 +504,6 @@ func tryDecodeBase64(s string) string {
return s return s
} }
// isPrintableText checks if a string contains mostly printable characters
func isPrintableText(s string) bool { func isPrintableText(s string) bool {
printableCount := 0 printableCount := 0
for _, r := range s { for _, r := range s {
@ -604,9 +524,6 @@ func showHelp() {
fmt.Printf(" status - Show network status\n") fmt.Printf(" status - Show network status\n")
fmt.Printf(" peer-id - Show this node's peer ID\n") fmt.Printf(" peer-id - Show this node's peer ID\n")
fmt.Printf(" query <sql> 🔐 Execute database query\n") fmt.Printf(" query <sql> 🔐 Execute database query\n")
fmt.Printf(" storage get <key> 🔐 Get value from storage\n")
fmt.Printf(" storage put <key> <value> 🔐 Store value in storage\n")
fmt.Printf(" storage list [prefix] 🔐 List storage keys\n")
fmt.Printf(" pubsub publish <topic> <msg> 🔐 Publish message\n") fmt.Printf(" pubsub publish <topic> <msg> 🔐 Publish message\n")
fmt.Printf(" pubsub subscribe <topic> [duration] 🔐 Subscribe to topic\n") fmt.Printf(" pubsub subscribe <topic> [duration] 🔐 Subscribe to topic\n")
fmt.Printf(" pubsub topics 🔐 List topics\n") fmt.Printf(" pubsub topics 🔐 List topics\n")
@ -628,12 +545,9 @@ func showHelp() {
fmt.Printf(" network-cli peer-id --format json\n") fmt.Printf(" network-cli peer-id --format json\n")
fmt.Printf(" network-cli peers --format json\n") fmt.Printf(" network-cli peers --format json\n")
fmt.Printf(" network-cli peers --production\n") fmt.Printf(" network-cli peers --production\n")
fmt.Printf(" network-cli storage put user:123 '{\"name\":\"Alice\"}'\n") fmt.Printf(" ./bin/network-cli pubsub publish notifications \"Hello World\"\n")
fmt.Printf(" network-cli pubsub subscribe notifications 1m\n")
} }
// Print functions
func printHealth(health *client.HealthStatus) { func printHealth(health *client.HealthStatus) {
fmt.Printf("🏥 Network Health\n") fmt.Printf("🏥 Network Health\n")
fmt.Printf("Status: %s\n", getStatusEmoji(health.Status)+health.Status) fmt.Printf("Status: %s\n", getStatusEmoji(health.Status)+health.Status)

512
e2e/gateway_e2e_test.go
View File

@ -3,11 +3,10 @@
package e2e package e2e
import ( import (
"crypto/rand" "crypto/rand"
"encoding/base64" "encoding/base64"
"encoding/json" "encoding/json"
"fmt" "fmt"
"io"
"net/http" "net/http"
"net/url" "net/url"
"os" "os"
@ -68,86 +67,6 @@ func TestGateway_Health(t *testing.T) {
} }
} }
func TestGateway_Storage_PutGetListExistsDelete(t *testing.T) {
key := requireAPIKey(t)
base := gatewayBaseURL()
// Unique key and payload
ts := time.Now().UnixNano()
kvKey := fmt.Sprintf("e2e-gw/%d", ts)
payload := randomBytes(32)
// PUT
{
req, err := http.NewRequest(http.MethodPost, base+"/v1/storage/put?key="+url.QueryEscape(kvKey), strings.NewReader(string(payload)))
if err != nil { t.Fatalf("put new req: %v", err) }
req.Header = authHeader(key)
resp, err := httpClient().Do(req)
if err != nil { t.Fatalf("put do: %v", err) }
defer resp.Body.Close()
if resp.StatusCode != http.StatusCreated {
t.Fatalf("put status: %d", resp.StatusCode)
}
}
// EXISTS
{
req, _ := http.NewRequest(http.MethodGet, base+"/v1/storage/exists?key="+url.QueryEscape(kvKey), nil)
req.Header = authHeader(key)
resp, err := httpClient().Do(req)
if err != nil { t.Fatalf("exists do: %v", err) }
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { t.Fatalf("exists status: %d", resp.StatusCode) }
var b struct{ Exists bool `json:"exists"` }
if err := json.NewDecoder(resp.Body).Decode(&b); err != nil { t.Fatalf("exists decode: %v", err) }
if !b.Exists { t.Fatalf("exists=false for %s", kvKey) }
}
// GET
{
req, _ := http.NewRequest(http.MethodGet, base+"/v1/storage/get?key="+url.QueryEscape(kvKey), nil)
req.Header = authHeader(key)
resp, err := httpClient().Do(req)
if err != nil { t.Fatalf("get do: %v", err) }
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { t.Fatalf("get status: %d", resp.StatusCode) }
got, _ := io.ReadAll(resp.Body)
if string(got) != string(payload) {
// Some deployments may base64-encode binary; accept if it decodes to the original
dec, derr := base64.StdEncoding.DecodeString(string(got))
if derr != nil || string(dec) != string(payload) {
t.Fatalf("payload mismatch: want %q got %q", string(payload), string(got))
}
}
}
// LIST (prefix)
{
prefix := url.QueryEscape(strings.Split(kvKey, "/")[0])
req, _ := http.NewRequest(http.MethodGet, base+"/v1/storage/list?prefix="+prefix, nil)
req.Header = authHeader(key)
resp, err := httpClient().Do(req)
if err != nil { t.Fatalf("list do: %v", err) }
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { t.Fatalf("list status: %d", resp.StatusCode) }
var b struct{ Keys []string `json:"keys"` }
if err := json.NewDecoder(resp.Body).Decode(&b); err != nil { t.Fatalf("list decode: %v", err) }
found := false
for _, k := range b.Keys { if k == kvKey { found = true; break } }
if !found { t.Fatalf("key %s not found in list", kvKey) }
}
// DELETE
{
req, _ := http.NewRequest(http.MethodPost, base+"/v1/storage/delete?key="+url.QueryEscape(kvKey), nil)
req.Header = authHeader(key)
resp, err := httpClient().Do(req)
if err != nil { t.Fatalf("delete do: %v", err) }
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { t.Fatalf("delete status: %d", resp.StatusCode) }
}
}
func TestGateway_PubSub_WS_Echo(t *testing.T) { func TestGateway_PubSub_WS_Echo(t *testing.T) {
key := requireAPIKey(t) key := requireAPIKey(t)
base := gatewayBaseURL() base := gatewayBaseURL()
@ -169,8 +88,12 @@ func TestGateway_PubSub_WS_Echo(t *testing.T) {
} }
_, data, err := c.ReadMessage() _, data, err := c.ReadMessage()
if err != nil { t.Fatalf("ws read: %v", err) } if err != nil {
if string(data) != string(msg) { t.Fatalf("ws echo mismatch: %q", string(data)) } t.Fatalf("ws read: %v", err)
}
if string(data) != string(msg) {
t.Fatalf("ws echo mismatch: %q", string(data))
}
} }
func TestGateway_PubSub_RestPublishToWS(t *testing.T) { func TestGateway_PubSub_RestPublishToWS(t *testing.T) {
@ -181,7 +104,9 @@ func TestGateway_PubSub_RestPublishToWS(t *testing.T) {
wsURL, hdr := toWSURL(base+"/v1/pubsub/ws?topic="+url.QueryEscape(topic)), http.Header{} wsURL, hdr := toWSURL(base+"/v1/pubsub/ws?topic="+url.QueryEscape(topic)), http.Header{}
hdr.Set("Authorization", "Bearer "+key) hdr.Set("Authorization", "Bearer "+key)
c, _, err := websocket.DefaultDialer.Dial(wsURL, hdr) c, _, err := websocket.DefaultDialer.Dial(wsURL, hdr)
if err != nil { t.Fatalf("ws dial: %v", err) } if err != nil {
t.Fatalf("ws dial: %v", err)
}
defer c.Close() defer c.Close()
// Publish via REST // Publish via REST
@ -191,198 +116,307 @@ func TestGateway_PubSub_RestPublishToWS(t *testing.T) {
req, _ := http.NewRequest(http.MethodPost, base+"/v1/pubsub/publish", strings.NewReader(body)) req, _ := http.NewRequest(http.MethodPost, base+"/v1/pubsub/publish", strings.NewReader(body))
req.Header = authHeader(key) req.Header = authHeader(key)
resp, err := httpClient().Do(req) resp, err := httpClient().Do(req)
if err != nil { t.Fatalf("publish do: %v", err) } if err != nil {
t.Fatalf("publish do: %v", err)
}
resp.Body.Close() resp.Body.Close()
if resp.StatusCode != http.StatusOK { t.Fatalf("publish status: %d", resp.StatusCode) } if resp.StatusCode != http.StatusOK {
t.Fatalf("publish status: %d", resp.StatusCode)
}
// Expect the message via WS // Expect the message via WS
_ = c.SetReadDeadline(time.Now().Add(5 * time.Second)) _ = c.SetReadDeadline(time.Now().Add(5 * time.Second))
_, data, err := c.ReadMessage() _, data, err := c.ReadMessage()
if err != nil { t.Fatalf("ws read: %v", err) } if err != nil {
if string(data) != string(payload) { t.Fatalf("payload mismatch: %q != %q", string(data), string(payload)) } t.Fatalf("ws read: %v", err)
}
if string(data) != string(payload) {
t.Fatalf("payload mismatch: %q != %q", string(data), string(payload))
}
// Topics list should include our topic (without namespace prefix) // Topics list should include our topic (without namespace prefix)
req2, _ := http.NewRequest(http.MethodGet, base+"/v1/pubsub/topics", nil) req2, _ := http.NewRequest(http.MethodGet, base+"/v1/pubsub/topics", nil)
req2.Header = authHeader(key) req2.Header = authHeader(key)
resp2, err := httpClient().Do(req2) resp2, err := httpClient().Do(req2)
if err != nil { t.Fatalf("topics do: %v", err) } if err != nil {
t.Fatalf("topics do: %v", err)
}
defer resp2.Body.Close() defer resp2.Body.Close()
if resp2.StatusCode != http.StatusOK { t.Fatalf("topics status: %d", resp2.StatusCode) } if resp2.StatusCode != http.StatusOK {
var tlist struct{ Topics []string `json:"topics"` } t.Fatalf("topics status: %d", resp2.StatusCode)
if err := json.NewDecoder(resp2.Body).Decode(&tlist); err != nil { t.Fatalf("topics decode: %v", err) } }
var tlist struct {
Topics []string `json:"topics"`
}
if err := json.NewDecoder(resp2.Body).Decode(&tlist); err != nil {
t.Fatalf("topics decode: %v", err)
}
found := false found := false
for _, tt := range tlist.Topics { if tt == topic { found = true; break } } for _, tt := range tlist.Topics {
if !found { t.Fatalf("topic %s not found in topics list", topic) } if tt == topic {
found = true
break
}
}
if !found {
t.Fatalf("topic %s not found in topics list", topic)
}
} }
func TestGateway_Database_CreateQueryMigrate(t *testing.T) { func TestGateway_Database_CreateQueryMigrate(t *testing.T) {
key := requireAPIKey(t) key := requireAPIKey(t)
base := gatewayBaseURL() base := gatewayBaseURL()
// Create table // Create table
schema := `CREATE TABLE IF NOT EXISTS e2e_items (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP)` schema := `CREATE TABLE IF NOT EXISTS e2e_items (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP)`
body := fmt.Sprintf(`{"schema":%q}`, schema) body := fmt.Sprintf(`{"schema":%q}`, schema)
req, _ := http.NewRequest(http.MethodPost, base+"/v1/db/create-table", strings.NewReader(body)) req, _ := http.NewRequest(http.MethodPost, base+"/v1/db/create-table", strings.NewReader(body))
req.Header = authHeader(key) req.Header = authHeader(key)
resp, err := httpClient().Do(req) resp, err := httpClient().Do(req)
if err != nil { t.Fatalf("create-table do: %v", err) } if err != nil {
resp.Body.Close() t.Fatalf("create-table do: %v", err)
if resp.StatusCode != http.StatusCreated { t.Fatalf("create-table status: %d", resp.StatusCode) } }
resp.Body.Close()
if resp.StatusCode != http.StatusCreated {
t.Fatalf("create-table status: %d", resp.StatusCode)
}
// Insert via transaction (simulate migration/data seed) // Insert via transaction (simulate migration/data seed)
txBody := `{"statements":["INSERT INTO e2e_items(name) VALUES ('one')","INSERT INTO e2e_items(name) VALUES ('two')"]}` txBody := `{"statements":["INSERT INTO e2e_items(name) VALUES ('one')","INSERT INTO e2e_items(name) VALUES ('two')"]}`
req, _ = http.NewRequest(http.MethodPost, base+"/v1/db/transaction", strings.NewReader(txBody)) req, _ = http.NewRequest(http.MethodPost, base+"/v1/db/transaction", strings.NewReader(txBody))
req.Header = authHeader(key) req.Header = authHeader(key)
resp, err = httpClient().Do(req) resp, err = httpClient().Do(req)
if err != nil { t.Fatalf("tx do: %v", err) } if err != nil {
resp.Body.Close() t.Fatalf("tx do: %v", err)
if resp.StatusCode != http.StatusOK { t.Fatalf("tx status: %d", resp.StatusCode) } }
resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("tx status: %d", resp.StatusCode)
}
// Query rows // Query rows
qBody := `{"sql":"SELECT name FROM e2e_items ORDER BY id ASC"}` qBody := `{"sql":"SELECT name FROM e2e_items ORDER BY id ASC"}`
req, _ = http.NewRequest(http.MethodPost, base+"/v1/db/query", strings.NewReader(qBody)) req, _ = http.NewRequest(http.MethodPost, base+"/v1/db/query", strings.NewReader(qBody))
req.Header = authHeader(key) req.Header = authHeader(key)
resp, err = httpClient().Do(req) resp, err = httpClient().Do(req)
if err != nil { t.Fatalf("query do: %v", err) } if err != nil {
defer resp.Body.Close() t.Fatalf("query do: %v", err)
if resp.StatusCode != http.StatusOK { t.Fatalf("query status: %d", resp.StatusCode) } }
var qr struct { Columns []string `json:"columns"`; Rows [][]any `json:"rows"`; Count int `json:"count"` } defer resp.Body.Close()
if err := json.NewDecoder(resp.Body).Decode(&qr); err != nil { t.Fatalf("query decode: %v", err) } if resp.StatusCode != http.StatusOK {
if qr.Count < 2 { t.Fatalf("expected at least 2 rows, got %d", qr.Count) } t.Fatalf("query status: %d", resp.StatusCode)
}
var qr struct {
Columns []string `json:"columns"`
Rows [][]any `json:"rows"`
Count int `json:"count"`
}
if err := json.NewDecoder(resp.Body).Decode(&qr); err != nil {
t.Fatalf("query decode: %v", err)
}
if qr.Count < 2 {
t.Fatalf("expected at least 2 rows, got %d", qr.Count)
}
// Schema endpoint returns tables // Schema endpoint returns tables
req, _ = http.NewRequest(http.MethodGet, base+"/v1/db/schema", nil) req, _ = http.NewRequest(http.MethodGet, base+"/v1/db/schema", nil)
req.Header = authHeader(key) req.Header = authHeader(key)
resp2, err := httpClient().Do(req) resp2, err := httpClient().Do(req)
if err != nil { t.Fatalf("schema do: %v", err) } if err != nil {
defer resp2.Body.Close() t.Fatalf("schema do: %v", err)
if resp2.StatusCode != http.StatusOK { t.Fatalf("schema status: %d", resp2.StatusCode) } }
defer resp2.Body.Close()
if resp2.StatusCode != http.StatusOK {
t.Fatalf("schema status: %d", resp2.StatusCode)
}
} }
func TestGateway_Database_DropTable(t *testing.T) { func TestGateway_Database_DropTable(t *testing.T) {
key := requireAPIKey(t) key := requireAPIKey(t)
base := gatewayBaseURL() base := gatewayBaseURL()
table := fmt.Sprintf("e2e_tmp_%d", time.Now().UnixNano()) table := fmt.Sprintf("e2e_tmp_%d", time.Now().UnixNano())
schema := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INTEGER PRIMARY KEY, note TEXT)", table) schema := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INTEGER PRIMARY KEY, note TEXT)", table)
// create // create
body := fmt.Sprintf(`{"schema":%q}`, schema) body := fmt.Sprintf(`{"schema":%q}`, schema)
req, _ := http.NewRequest(http.MethodPost, base+"/v1/db/create-table", strings.NewReader(body)) req, _ := http.NewRequest(http.MethodPost, base+"/v1/db/create-table", strings.NewReader(body))
req.Header = authHeader(key) req.Header = authHeader(key)
resp, err := httpClient().Do(req) resp, err := httpClient().Do(req)
if err != nil { t.Fatalf("create-table do: %v", err) } if err != nil {
resp.Body.Close() t.Fatalf("create-table do: %v", err)
if resp.StatusCode != http.StatusCreated { t.Fatalf("create-table status: %d", resp.StatusCode) } }
// drop resp.Body.Close()
dbody := fmt.Sprintf(`{"table":%q}`, table) if resp.StatusCode != http.StatusCreated {
req, _ = http.NewRequest(http.MethodPost, base+"/v1/db/drop-table", strings.NewReader(dbody)) t.Fatalf("create-table status: %d", resp.StatusCode)
req.Header = authHeader(key) }
resp, err = httpClient().Do(req) // drop
if err != nil { t.Fatalf("drop-table do: %v", err) } dbody := fmt.Sprintf(`{"table":%q}`, table)
resp.Body.Close() req, _ = http.NewRequest(http.MethodPost, base+"/v1/db/drop-table", strings.NewReader(dbody))
if resp.StatusCode != http.StatusOK { t.Fatalf("drop-table status: %d", resp.StatusCode) } req.Header = authHeader(key)
// verify not in schema resp, err = httpClient().Do(req)
req, _ = http.NewRequest(http.MethodGet, base+"/v1/db/schema", nil) if err != nil {
req.Header = authHeader(key) t.Fatalf("drop-table do: %v", err)
resp2, err := httpClient().Do(req) }
if err != nil { t.Fatalf("schema do: %v", err) } resp.Body.Close()
defer resp2.Body.Close() if resp.StatusCode != http.StatusOK {
if resp2.StatusCode != http.StatusOK { t.Fatalf("schema status: %d", resp2.StatusCode) } t.Fatalf("drop-table status: %d", resp.StatusCode)
var schemaResp struct{ Tables []struct{ Name string `json:"name"` } `json:"tables"` } }
if err := json.NewDecoder(resp2.Body).Decode(&schemaResp); err != nil { t.Fatalf("schema decode: %v", err) } // verify not in schema
for _, tbl := range schemaResp.Tables { if tbl.Name == table { t.Fatalf("table %s still present after drop", table) } } req, _ = http.NewRequest(http.MethodGet, base+"/v1/db/schema", nil)
req.Header = authHeader(key)
resp2, err := httpClient().Do(req)
if err != nil {
t.Fatalf("schema do: %v", err)
}
defer resp2.Body.Close()
if resp2.StatusCode != http.StatusOK {
t.Fatalf("schema status: %d", resp2.StatusCode)
}
var schemaResp struct {
Tables []struct {
Name string `json:"name"`
} `json:"tables"`
}
if err := json.NewDecoder(resp2.Body).Decode(&schemaResp); err != nil {
t.Fatalf("schema decode: %v", err)
}
for _, tbl := range schemaResp.Tables {
if tbl.Name == table {
t.Fatalf("table %s still present after drop", table)
}
}
} }
func TestGateway_Database_RecreateWithFK(t *testing.T) { func TestGateway_Database_RecreateWithFK(t *testing.T) {
key := requireAPIKey(t) key := requireAPIKey(t)
base := gatewayBaseURL() base := gatewayBaseURL()
// base tables // base tables
orgs := fmt.Sprintf("e2e_orgs_%d", time.Now().UnixNano()) orgs := fmt.Sprintf("e2e_orgs_%d", time.Now().UnixNano())
users := fmt.Sprintf("e2e_users_%d", time.Now().UnixNano()) users := fmt.Sprintf("e2e_users_%d", time.Now().UnixNano())
createOrgs := fmt.Sprintf(`{"schema":%q}`, fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INTEGER PRIMARY KEY, name TEXT)", orgs)) createOrgs := fmt.Sprintf(`{"schema":%q}`, fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INTEGER PRIMARY KEY, name TEXT)", orgs))
createUsers := fmt.Sprintf(`{"schema":%q}`, fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INTEGER PRIMARY KEY, name TEXT, org_id INTEGER, age TEXT)", users)) createUsers := fmt.Sprintf(`{"schema":%q}`, fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INTEGER PRIMARY KEY, name TEXT, org_id INTEGER, age TEXT)", users))
for _, body := range []string{createOrgs, createUsers} { for _, body := range []string{createOrgs, createUsers} {
req, _ := http.NewRequest(http.MethodPost, base+"/v1/db/create-table", strings.NewReader(body)) req, _ := http.NewRequest(http.MethodPost, base+"/v1/db/create-table", strings.NewReader(body))
req.Header = authHeader(key) req.Header = authHeader(key)
resp, err := httpClient().Do(req) resp, err := httpClient().Do(req)
if err != nil { t.Fatalf("create-table do: %v", err) } if err != nil {
resp.Body.Close() t.Fatalf("create-table do: %v", err)
if resp.StatusCode != http.StatusCreated { t.Fatalf("create-table status: %d", resp.StatusCode) } }
} resp.Body.Close()
// seed data if resp.StatusCode != http.StatusCreated {
txSeed := fmt.Sprintf(`{"statements":["INSERT INTO %s(id,name) VALUES (1,'org')","INSERT INTO %s(id,name,org_id,age) VALUES (1,'alice',1,'30')"]}`, orgs, users) t.Fatalf("create-table status: %d", resp.StatusCode)
req, _ := http.NewRequest(http.MethodPost, base+"/v1/db/transaction", strings.NewReader(txSeed)) }
req.Header = authHeader(key) }
resp, err := httpClient().Do(req) // seed data
if err != nil { t.Fatalf("seed tx do: %v", err) } txSeed := fmt.Sprintf(`{"statements":["INSERT INTO %s(id,name) VALUES (1,'org')","INSERT INTO %s(id,name,org_id,age) VALUES (1,'alice',1,'30')"]}`, orgs, users)
resp.Body.Close() req, _ := http.NewRequest(http.MethodPost, base+"/v1/db/transaction", strings.NewReader(txSeed))
if resp.StatusCode != http.StatusOK { t.Fatalf("seed tx status: %d", resp.StatusCode) } req.Header = authHeader(key)
resp, err := httpClient().Do(req)
if err != nil {
t.Fatalf("seed tx do: %v", err)
}
resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("seed tx status: %d", resp.StatusCode)
}
// migrate: change users.age TEXT -> INTEGER and add FK to orgs(id) // migrate: change users.age TEXT -> INTEGER and add FK to orgs(id)
// Note: Some backends may not support connection-scoped BEGIN/COMMIT or PRAGMA via HTTP. // Note: Some backends may not support connection-scoped BEGIN/COMMIT or PRAGMA via HTTP.
// We apply the standard recreate pattern without explicit PRAGMAs/transaction. // We apply the standard recreate pattern without explicit PRAGMAs/transaction.
txMig := fmt.Sprintf(`{"statements":[ txMig := fmt.Sprintf(`{"statements":[
"CREATE TABLE %s_new (id INTEGER PRIMARY KEY, name TEXT, org_id INTEGER, age INTEGER, FOREIGN KEY(org_id) REFERENCES %s(id) ON DELETE CASCADE)", "CREATE TABLE %s_new (id INTEGER PRIMARY KEY, name TEXT, org_id INTEGER, age INTEGER, FOREIGN KEY(org_id) REFERENCES %s(id) ON DELETE CASCADE)",
"INSERT INTO %s_new (id,name,org_id,age) SELECT id,name,org_id, CAST(age AS INTEGER) FROM %s", "INSERT INTO %s_new (id,name,org_id,age) SELECT id,name,org_id, CAST(age AS INTEGER) FROM %s",
"DROP TABLE %s", "DROP TABLE %s",
"ALTER TABLE %s_new RENAME TO %s" "ALTER TABLE %s_new RENAME TO %s"
]}` , users, orgs, users, users, users, users, users) ]}`, users, orgs, users, users, users, users, users)
req, _ = http.NewRequest(http.MethodPost, base+"/v1/db/transaction", strings.NewReader(txMig)) req, _ = http.NewRequest(http.MethodPost, base+"/v1/db/transaction", strings.NewReader(txMig))
req.Header = authHeader(key) req.Header = authHeader(key)
resp, err = httpClient().Do(req) resp, err = httpClient().Do(req)
if err != nil { t.Fatalf("mig tx do: %v", err) } if err != nil {
resp.Body.Close() t.Fatalf("mig tx do: %v", err)
if resp.StatusCode != http.StatusOK { t.Fatalf("mig tx status: %d", resp.StatusCode) } }
resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("mig tx status: %d", resp.StatusCode)
}
// verify schema type change // verify schema type change
qBody := fmt.Sprintf(`{"sql":"PRAGMA table_info(%s)"}`, users) qBody := fmt.Sprintf(`{"sql":"PRAGMA table_info(%s)"}`, users)
req, _ = http.NewRequest(http.MethodPost, base+"/v1/db/query", strings.NewReader(qBody)) req, _ = http.NewRequest(http.MethodPost, base+"/v1/db/query", strings.NewReader(qBody))
req.Header = authHeader(key) req.Header = authHeader(key)
resp, err = httpClient().Do(req) resp, err = httpClient().Do(req)
if err != nil { t.Fatalf("pragma do: %v", err) } if err != nil {
defer resp.Body.Close() t.Fatalf("pragma do: %v", err)
if resp.StatusCode != http.StatusOK { t.Fatalf("pragma status: %d", resp.StatusCode) } }
var qr struct{ Columns []string `json:"columns"`; Rows [][]any `json:"rows"` } defer resp.Body.Close()
if err := json.NewDecoder(resp.Body).Decode(&qr); err != nil { t.Fatalf("pragma decode: %v", err) } if resp.StatusCode != http.StatusOK {
// column order: cid,name,type,notnull,dflt_value,pk t.Fatalf("pragma status: %d", resp.StatusCode)
ageIsInt := false }
for _, row := range qr.Rows { var qr struct {
if len(row) >= 3 && fmt.Sprintf("%v", row[1]) == "age" { Columns []string `json:"columns"`
tstr := strings.ToUpper(fmt.Sprintf("%v", row[2])) Rows [][]any `json:"rows"`
if strings.Contains(tstr, "INT") { ageIsInt = true; break } }
} if err := json.NewDecoder(resp.Body).Decode(&qr); err != nil {
} t.Fatalf("pragma decode: %v", err)
if !ageIsInt { }
// Fallback: inspect CREATE TABLE SQL from sqlite_master // column order: cid,name,type,notnull,dflt_value,pk
qBody2 := fmt.Sprintf(`{"sql":"SELECT sql FROM sqlite_master WHERE type='table' AND name='%s'"}`, users) ageIsInt := false
req2, _ := http.NewRequest(http.MethodPost, base+"/v1/db/query", strings.NewReader(qBody2)) for _, row := range qr.Rows {
req2.Header = authHeader(key) if len(row) >= 3 && fmt.Sprintf("%v", row[1]) == "age" {
resp3, err := httpClient().Do(req2) tstr := strings.ToUpper(fmt.Sprintf("%v", row[2]))
if err != nil { t.Fatalf("sqlite_master do: %v", err) } if strings.Contains(tstr, "INT") {
defer resp3.Body.Close() ageIsInt = true
if resp3.StatusCode != http.StatusOK { t.Fatalf("sqlite_master status: %d", resp3.StatusCode) } break
var qr2 struct{ Rows [][]any `json:"rows"` } }
if err := json.NewDecoder(resp3.Body).Decode(&qr2); err != nil { t.Fatalf("sqlite_master decode: %v", err) } }
found := false }
for _, row := range qr2.Rows { if !ageIsInt {
if len(row) > 0 { // Fallback: inspect CREATE TABLE SQL from sqlite_master
sql := strings.ToUpper(fmt.Sprintf("%v", row[0])) qBody2 := fmt.Sprintf(`{"sql":"SELECT sql FROM sqlite_master WHERE type='table' AND name='%s'"}`, users)
if strings.Contains(sql, "AGE INT") || strings.Contains(sql, "AGE INTEGER") { req2, _ := http.NewRequest(http.MethodPost, base+"/v1/db/query", strings.NewReader(qBody2))
found = true req2.Header = authHeader(key)
break resp3, err := httpClient().Do(req2)
} if err != nil {
} t.Fatalf("sqlite_master do: %v", err)
} }
if !found { t.Fatalf("age column type not INTEGER after migration") } defer resp3.Body.Close()
} if resp3.StatusCode != http.StatusOK {
t.Fatalf("sqlite_master status: %d", resp3.StatusCode)
}
var qr2 struct {
Rows [][]any `json:"rows"`
}
if err := json.NewDecoder(resp3.Body).Decode(&qr2); err != nil {
t.Fatalf("sqlite_master decode: %v", err)
}
found := false
for _, row := range qr2.Rows {
if len(row) > 0 {
sql := strings.ToUpper(fmt.Sprintf("%v", row[0]))
if strings.Contains(sql, "AGE INT") || strings.Contains(sql, "AGE INTEGER") {
found = true
break
}
}
}
if !found {
t.Fatalf("age column type not INTEGER after migration")
}
}
} }
func toWSURL(httpURL string) string { func toWSURL(httpURL string) string {
u, err := url.Parse(httpURL) u, err := url.Parse(httpURL)
if err != nil { return httpURL } if err != nil {
if u.Scheme == "https" { u.Scheme = "wss" } else { u.Scheme = "ws" } return httpURL
}
if u.Scheme == "https" {
u.Scheme = "wss"
} else {
u.Scheme = "ws"
}
return u.String() return u.String()
} }

44
examples/basic_usage.go
View File

@ -32,9 +32,6 @@ func main() {
// Example: Database operations // Example: Database operations
demonstrateDatabase(networkClient) demonstrateDatabase(networkClient)
// Example: Storage operations
demonstrateStorage(networkClient)
// Example: Pub/Sub messaging // Example: Pub/Sub messaging
demonstratePubSub(networkClient) demonstratePubSub(networkClient)
@ -83,47 +80,6 @@ func demonstrateDatabase(client client.NetworkClient) {
log.Printf("Query result: %+v", result) log.Printf("Query result: %+v", result)
} }
func demonstrateStorage(client client.NetworkClient) {
ctx := context.Background()
storage := client.Storage()
log.Printf("=== Storage Operations ===")
// Store some data
key := "user:123"
value := []byte(`{"name": "Alice", "age": 30}`)
if err := storage.Put(ctx, key, value); err != nil {
log.Printf("Error storing data: %v", err)
return
}
log.Printf("Data stored successfully")
// Retrieve data
retrieved, err := storage.Get(ctx, key)
if err != nil {
log.Printf("Error retrieving data: %v", err)
return
}
log.Printf("Retrieved data: %s", string(retrieved))
// Check if key exists
exists, err := storage.Exists(ctx, key)
if err != nil {
log.Printf("Error checking existence: %v", err)
return
}
log.Printf("Key exists: %v", exists)
// List keys
keys, err := storage.List(ctx, "user:", 10)
if err != nil {
log.Printf("Error listing keys: %v", err)
return
}
log.Printf("Keys: %v", keys)
}
func demonstratePubSub(client client.NetworkClient) { func demonstratePubSub(client client.NetworkClient) {
ctx := context.Background() ctx := context.Background()
pubsub := client.PubSub() pubsub := client.PubSub()

6555
network.md Normal file
View File

File diff suppressed because one or more lines are too long

23
pkg/client/client.go
View File

@ -22,7 +22,6 @@ import (
"git.debros.io/DeBros/network/pkg/anyoneproxy" "git.debros.io/DeBros/network/pkg/anyoneproxy"
"git.debros.io/DeBros/network/pkg/pubsub" "git.debros.io/DeBros/network/pkg/pubsub"
"git.debros.io/DeBros/network/pkg/storage"
) )
// Client implements the NetworkClient interface // Client implements the NetworkClient interface
@ -36,7 +35,6 @@ type Client struct {
// Components // Components
database *DatabaseClientImpl database *DatabaseClientImpl
storage *StorageClientImpl
network *NetworkInfoImpl network *NetworkInfoImpl
pubsub *pubSubBridge pubsub *pubSubBridge
@ -83,11 +81,6 @@ func (c *Client) Database() DatabaseClient {
return c.database return c.database
} }
// Storage returns the storage client
func (c *Client) Storage() StorageClient {
return c.storage
}
// PubSub returns the pub/sub client // PubSub returns the pub/sub client
func (c *Client) PubSub() PubSubClient { func (c *Client) PubSub() PubSubClient {
return c.pubsub return c.pubsub
@ -202,16 +195,6 @@ func (c *Client) Connect() error {
c.pubsub = &pubSubBridge{client: c, adapter: adapter} c.pubsub = &pubSubBridge{client: c, adapter: adapter}
c.logger.Info("Pubsub bridge created successfully") c.logger.Info("Pubsub bridge created successfully")
c.logger.Info("Creating storage client...")
// Create storage client with the host (use namespace directly to avoid deadlock)
storageClient := storage.NewClient(h, namespace, c.logger)
c.storage = &StorageClientImpl{
client: c,
storageClient: storageClient,
}
c.logger.Info("Storage client created successfully")
c.logger.Info("Starting bootstrap peer connections...") c.logger.Info("Starting bootstrap peer connections...")
// Connect to bootstrap peers FIRST // Connect to bootstrap peers FIRST
@ -315,7 +298,6 @@ func (c *Client) Health() (*HealthStatus, error) {
checks := map[string]string{ checks := map[string]string{
"connection": "ok", "connection": "ok",
"database": "ok", "database": "ok",
"storage": "ok",
"pubsub": "ok", "pubsub": "ok",
} }
@ -360,11 +342,6 @@ func (c *Client) requireAccess(ctx context.Context) error {
return fmt.Errorf("access denied: API key or JWT required") return fmt.Errorf("access denied: API key or JWT required")
} }
ns := c.getAppNamespace() ns := c.getAppNamespace()
if v := ctx.Value(storage.CtxKeyNamespaceOverride); v != nil {
if s, ok := v.(string); ok && s != "" && s != ns {
return fmt.Errorf("access denied: namespace mismatch")
}
}
if v := ctx.Value(pubsub.CtxKeyNamespaceOverride); v != nil { if v := ctx.Value(pubsub.CtxKeyNamespaceOverride); v != nil {
if s, ok := v.(string); ok && s != "" && s != ns { if s, ok := v.(string); ok && s != "" && s != ns {
return fmt.Errorf("access denied: namespace mismatch") return fmt.Errorf("access denied: namespace mismatch")

7
pkg/client/client_test.go
View File

@ -7,7 +7,6 @@ import (
"testing" "testing"
"git.debros.io/DeBros/network/pkg/pubsub" "git.debros.io/DeBros/network/pkg/pubsub"
"git.debros.io/DeBros/network/pkg/storage"
) )
// MakeJWT creates a minimal JWT-like token with a json payload // MakeJWT creates a minimal JWT-like token with a json payload
@ -146,10 +145,10 @@ func TestRequireAccess(t *testing.T) {
// set resolved namespace to "app" to simulate derived namespace // set resolved namespace to "app" to simulate derived namespace
c.resolvedNamespace = "app" c.resolvedNamespace = "app"
// override storage namespace to something else // override pubsub namespace to something else
ctx := storage.WithNamespace(context.Background(), "other") ctx := pubsub.WithNamespace(context.Background(), "other")
if err := c.requireAccess(ctx); err == nil { if err := c.requireAccess(ctx); err == nil {
t.Fatalf("expected namespace mismatch error for storage override") t.Fatalf("expected namespace mismatch error for pubsub override")
} }
// override pubsub namespace to something else // override pubsub namespace to something else

6
pkg/client/context.go
View File

@ -4,7 +4,6 @@ import (
"context" "context"
"git.debros.io/DeBros/network/pkg/pubsub" "git.debros.io/DeBros/network/pkg/pubsub"
"git.debros.io/DeBros/network/pkg/storage"
) )
// contextKey for internal operations // contextKey for internal operations
@ -15,11 +14,10 @@ const (
ctxKeyInternal contextKey = "internal_operation" ctxKeyInternal contextKey = "internal_operation"
) )
// WithNamespace applies both storage and pubsub namespace overrides to the context. // WithNamespace applies pubsub namespace override to the context.
// It is a convenience helper for client callers to ensure both subsystems receive // It is a convenience helper for client callers to ensure subsystems receive
// the same, consistent namespace override. // the same, consistent namespace override.
func WithNamespace(ctx context.Context, ns string) context.Context { func WithNamespace(ctx context.Context, ns string) context.Context {
ctx = storage.WithNamespace(ctx, ns)
ctx = pubsub.WithNamespace(ctx, ns) ctx = pubsub.WithNamespace(ctx, ns)
return ctx return ctx
} }

83
pkg/client/implementations.go
View File

@ -7,8 +7,6 @@ import (
"sync" "sync"
"time" "time"
"git.debros.io/DeBros/network/pkg/storage"
"git.debros.io/DeBros/network/pkg/anyoneproxy" "git.debros.io/DeBros/network/pkg/anyoneproxy"
"github.com/libp2p/go-libp2p/core/peer" "github.com/libp2p/go-libp2p/core/peer"
"github.com/multiformats/go-multiaddr" "github.com/multiformats/go-multiaddr"
@ -400,87 +398,6 @@ func (d *DatabaseClientImpl) GetSchema(ctx context.Context) (*SchemaInfo, error)
return schema, nil return schema, nil
} }
// StorageClientImpl implements StorageClient using distributed storage
type StorageClientImpl struct {
client *Client
storageClient *storage.Client
}
// Get retrieves a value by key
func (s *StorageClientImpl) Get(ctx context.Context, key string) ([]byte, error) {
if !s.client.isConnected() {
return nil, fmt.Errorf("client not connected")
}
if err := s.client.requireAccess(ctx); err != nil {
return nil, fmt.Errorf("authentication required: %w - run CLI commands to authenticate automatically", err)
}
return s.storageClient.Get(ctx, key)
}
// Put stores a value by key
func (s *StorageClientImpl) Put(ctx context.Context, key string, value []byte) error {
if !s.client.isConnected() {
return fmt.Errorf("client not connected")
}
if err := s.client.requireAccess(ctx); err != nil {
return fmt.Errorf("authentication required: %w - run CLI commands to authenticate automatically", err)
}
err := s.storageClient.Put(ctx, key, value)
if err != nil {
return err
}
return nil
}
// Delete removes a key
func (s *StorageClientImpl) Delete(ctx context.Context, key string) error {
if !s.client.isConnected() {
return fmt.Errorf("client not connected")
}
if err := s.client.requireAccess(ctx); err != nil {
return fmt.Errorf("authentication required: %w - run CLI commands to authenticate automatically", err)
}
err := s.storageClient.Delete(ctx, key)
if err != nil {
return err
}
return nil
}
// List returns keys with a given prefix
func (s *StorageClientImpl) List(ctx context.Context, prefix string, limit int) ([]string, error) {
if !s.client.isConnected() {
return nil, fmt.Errorf("client not connected")
}
if err := s.client.requireAccess(ctx); err != nil {
return nil, fmt.Errorf("authentication required: %w - run CLI commands to authenticate automatically", err)
}
return s.storageClient.List(ctx, prefix, limit)
}
// Exists checks if a key exists
func (s *StorageClientImpl) Exists(ctx context.Context, key string) (bool, error) {
if !s.client.isConnected() {
return false, fmt.Errorf("client not connected")
}
if err := s.client.requireAccess(ctx); err != nil {
return false, fmt.Errorf("authentication required: %w - run CLI commands to authenticate automatically", err)
}
return s.storageClient.Exists(ctx, key)
}
// NetworkInfoImpl implements NetworkInfo // NetworkInfoImpl implements NetworkInfo
type NetworkInfoImpl struct { type NetworkInfoImpl struct {
client *Client client *Client

12
pkg/client/interface.go
View File

@ -11,9 +11,6 @@ type NetworkClient interface {
// Database operations (namespaced per app) // Database operations (namespaced per app)
Database() DatabaseClient Database() DatabaseClient
// Key-value storage (namespaced per app)
Storage() StorageClient
// Pub/Sub messaging // Pub/Sub messaging
PubSub() PubSubClient PubSub() PubSubClient
@ -38,15 +35,6 @@ type DatabaseClient interface {
GetSchema(ctx context.Context) (*SchemaInfo, error) GetSchema(ctx context.Context) (*SchemaInfo, error)
} }
// StorageClient provides key-value storage operations
type StorageClient interface {
Get(ctx context.Context, key string) ([]byte, error)
Put(ctx context.Context, key string, value []byte) error
Delete(ctx context.Context, key string) error
List(ctx context.Context, prefix string, limit int) ([]string, error)
Exists(ctx context.Context, key string) (bool, error)
}
// PubSubClient provides publish/subscribe messaging // PubSubClient provides publish/subscribe messaging
type PubSubClient interface { type PubSubClient interface {
Subscribe(ctx context.Context, topic string, handler MessageHandler) error Subscribe(ctx context.Context, topic string, handler MessageHandler) error

2
pkg/config/config.go
View File

@ -109,7 +109,7 @@ func DefaultConfig() *Config {
}, },
Discovery: DiscoveryConfig{ Discovery: DiscoveryConfig{
BootstrapPeers: []string{ BootstrapPeers: []string{
"/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWGqqR8bxgmYsYrGYMKnUWwZUCpioLmA3H37ggRDnAiFa7", "/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWSHHwEY6cga3ng7tD1rzStAU58ogQXVMX3LZJ6Gqf6dee",
}, },
BootstrapPort: 4001, // Default LibP2P port BootstrapPort: 4001, // Default LibP2P port
DiscoveryInterval: time.Second * 15, // Back to 15 seconds for testing DiscoveryInterval: time.Second * 15, // Back to 15 seconds for testing

8
pkg/gateway/routes.go
View File

@ -31,14 +31,6 @@ func (g *Gateway) Routes() http.Handler {
mux.HandleFunc("/v1/apps", g.appsHandler) mux.HandleFunc("/v1/apps", g.appsHandler)
mux.HandleFunc("/v1/apps/", g.appsHandler) mux.HandleFunc("/v1/apps/", g.appsHandler)
// storage
mux.HandleFunc("/v1/storage", g.storageHandler) // legacy/basic
mux.HandleFunc("/v1/storage/get", g.storageGetHandler)
mux.HandleFunc("/v1/storage/put", g.storagePutHandler)
mux.HandleFunc("/v1/storage/delete", g.storageDeleteHandler)
mux.HandleFunc("/v1/storage/list", g.storageListHandler)
mux.HandleFunc("/v1/storage/exists", g.storageExistsHandler)
// database // database
mux.HandleFunc("/v1/db/query", g.dbQueryHandler) mux.HandleFunc("/v1/db/query", g.dbQueryHandler)
mux.HandleFunc("/v1/db/transaction", g.dbTransactionHandler) mux.HandleFunc("/v1/db/transaction", g.dbTransactionHandler)

302
pkg/gateway/storage_handlers.go
View File

@ -2,118 +2,118 @@ package gateway
import ( import (
"encoding/json" "encoding/json"
"io"
"net/http" "net/http"
"strconv"
"git.debros.io/DeBros/network/pkg/client" "git.debros.io/DeBros/network/pkg/client"
"git.debros.io/DeBros/network/pkg/storage" "git.debros.io/DeBros/network/pkg/pubsub"
) )
// Database HTTP handlers // Database HTTP handlers
func (g *Gateway) dbQueryHandler(w http.ResponseWriter, r *http.Request) { func (g *Gateway) dbQueryHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil { writeError(w, http.StatusServiceUnavailable, "client not initialized"); return } if g.client == nil {
if r.Method != http.MethodPost { writeError(w, http.StatusMethodNotAllowed, "method not allowed"); return } writeError(w, http.StatusServiceUnavailable, "client not initialized");
var body struct{ SQL string `json:"sql"`; Args []any `json:"args"` } return
if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.SQL == "" { writeError(w, http.StatusBadRequest, "invalid body: {sql, args?}"); return } }
ctx := client.WithInternalAuth(r.Context()) if r.Method != http.MethodPost {
res, err := g.client.Database().Query(ctx, body.SQL, body.Args...) writeError(w, http.StatusMethodNotAllowed, "method not allowed");
if err != nil { writeError(w, http.StatusInternalServerError, err.Error()); return } return
writeJSON(w, http.StatusOK, res) }
var body struct{ SQL string `json:"sql"`; Args []any `json:"args"` }
if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.SQL == "" {
writeError(w, http.StatusBadRequest, "invalid body: {sql, args?}");
return
}
ctx := client.WithInternalAuth(r.Context())
res, err := g.client.Database().Query(ctx, body.SQL, body.Args...)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error());
return
}
writeJSON(w, http.StatusOK, res)
} }
func (g *Gateway) dbTransactionHandler(w http.ResponseWriter, r *http.Request) { func (g *Gateway) dbTransactionHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil { writeError(w, http.StatusServiceUnavailable, "client not initialized"); return } if g.client == nil {
if r.Method != http.MethodPost { writeError(w, http.StatusMethodNotAllowed, "method not allowed"); return } writeError(w, http.StatusServiceUnavailable, "client not initialized");
var body struct{ Statements []string `json:"statements"` } return
if err := json.NewDecoder(r.Body).Decode(&body); err != nil || len(body.Statements) == 0 { writeError(w, http.StatusBadRequest, "invalid body: {statements:[...]}"); return } }
ctx := client.WithInternalAuth(r.Context()) if r.Method != http.MethodPost {
if err := g.client.Database().Transaction(ctx, body.Statements); err != nil { writeError(w, http.StatusInternalServerError, err.Error()); return } writeError(w, http.StatusMethodNotAllowed, "method not allowed");
writeJSON(w, http.StatusOK, map[string]any{"status":"ok"}) return
}
var body struct{ Statements []string `json:"statements"` }
if err := json.NewDecoder(r.Body).Decode(&body); err != nil || len(body.Statements) == 0 {
writeError(w, http.StatusBadRequest, "invalid body: {statements:[...]}");
return
}
ctx := client.WithInternalAuth(r.Context())
if err := g.client.Database().Transaction(ctx, body.Statements); err != nil {
writeError(w, http.StatusInternalServerError, err.Error());
return
}
writeJSON(w, http.StatusOK, map[string]any{"status":"ok"})
} }
func (g *Gateway) dbSchemaHandler(w http.ResponseWriter, r *http.Request) { func (g *Gateway) dbSchemaHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil { writeError(w, http.StatusServiceUnavailable, "client not initialized"); return } if g.client == nil {
if r.Method != http.MethodGet { writeError(w, http.StatusMethodNotAllowed, "method not allowed"); return } writeError(w, http.StatusServiceUnavailable, "client not initialized");
ctx := client.WithInternalAuth(r.Context()) return
schema, err := g.client.Database().GetSchema(ctx) }
if err != nil { writeError(w, http.StatusInternalServerError, err.Error()); return } if r.Method != http.MethodGet {
writeJSON(w, http.StatusOK, schema) writeError(w, http.StatusMethodNotAllowed, "method not allowed");
return
}
ctx := client.WithInternalAuth(r.Context())
schema, err := g.client.Database().GetSchema(ctx)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error());
return
}
writeJSON(w, http.StatusOK, schema)
} }
func (g *Gateway) dbCreateTableHandler(w http.ResponseWriter, r *http.Request) { func (g *Gateway) dbCreateTableHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil { writeError(w, http.StatusServiceUnavailable, "client not initialized"); return } if g.client == nil {
if r.Method != http.MethodPost { writeError(w, http.StatusMethodNotAllowed, "method not allowed"); return } writeError(w, http.StatusServiceUnavailable, "client not initialized");
var body struct{ Schema string `json:"schema"` } return
if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.Schema == "" { writeError(w, http.StatusBadRequest, "invalid body: {schema}"); return } }
ctx := client.WithInternalAuth(r.Context()) if r.Method != http.MethodPost {
if err := g.client.Database().CreateTable(ctx, body.Schema); err != nil { writeError(w, http.StatusInternalServerError, err.Error()); return } writeError(w, http.StatusMethodNotAllowed, "method not allowed");
writeJSON(w, http.StatusCreated, map[string]any{"status":"ok"}) return
}
var body struct{ Schema string `json:"schema"` }
if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.Schema == "" {
writeError(w, http.StatusBadRequest, "invalid body: {schema}");
return
}
ctx := client.WithInternalAuth(r.Context())
if err := g.client.Database().CreateTable(ctx, body.Schema); err != nil {
writeError(w, http.StatusInternalServerError, err.Error());
return
}
writeJSON(w, http.StatusCreated, map[string]any{"status":"ok"})
} }
func (g *Gateway) dbDropTableHandler(w http.ResponseWriter, r *http.Request) { func (g *Gateway) dbDropTableHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil { writeError(w, http.StatusServiceUnavailable, "client not initialized"); return }
if r.Method != http.MethodPost { writeError(w, http.StatusMethodNotAllowed, "method not allowed"); return }
var body struct{ Table string `json:"table"` }
if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.Table == "" { writeError(w, http.StatusBadRequest, "invalid body: {table}"); return }
ctx := client.WithInternalAuth(r.Context())
if err := g.client.Database().DropTable(ctx, body.Table); err != nil { writeError(w, http.StatusInternalServerError, err.Error()); return }
writeJSON(w, http.StatusOK, map[string]any{"status":"ok"})
}
func (g *Gateway) storageHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil { if g.client == nil {
writeError(w, http.StatusServiceUnavailable, "client not initialized") writeError(w, http.StatusServiceUnavailable, "client not initialized");
return return
} }
if r.Method != http.MethodPost {
key := r.URL.Query().Get("key") writeError(w, http.StatusMethodNotAllowed, "method not allowed");
if key == "" {
writeError(w, http.StatusBadRequest, "missing 'key' query parameter")
return return
} }
var body struct{ Table string `json:"table"` }
// Use internal auth for downstream client calls; gateway has already authenticated the request if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.Table == "" {
ctx := client.WithInternalAuth(r.Context()) writeError(w, http.StatusBadRequest, "invalid body: {table}");
switch r.Method {
case http.MethodGet:
val, err := g.client.Storage().Get(ctx, key)
if err != nil {
// Some storage backends may return base64-encoded text; try best-effort decode for transparency
writeError(w, http.StatusNotFound, err.Error())
return
}
w.Header().Set("Content-Type", "application/octet-stream")
w.WriteHeader(http.StatusOK)
_, _ = w.Write(val)
return
case http.MethodPut:
defer r.Body.Close()
b, err := io.ReadAll(r.Body)
if err != nil {
writeError(w, http.StatusBadRequest, "failed to read body")
return
}
if err := g.client.Storage().Put(ctx, key, b); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusCreated, map[string]any{
"status": "ok",
"key": key,
"size": len(b),
})
return
case http.MethodOptions:
w.WriteHeader(http.StatusNoContent)
return
default:
writeError(w, http.StatusMethodNotAllowed, "method not allowed")
return return
} }
ctx := client.WithInternalAuth(r.Context())
if err := g.client.Database().DropTable(ctx, body.Table); err != nil {
writeError(w, http.StatusInternalServerError, err.Error());
return
}
writeJSON(w, http.StatusOK, map[string]any{"status":"ok"})
} }
func (g *Gateway) networkStatusHandler(w http.ResponseWriter, r *http.Request) { func (g *Gateway) networkStatusHandler(w http.ResponseWriter, r *http.Request) {
@ -144,134 +144,6 @@ func (g *Gateway) networkPeersHandler(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, peers) writeJSON(w, http.StatusOK, peers)
} }
func (g *Gateway) storageGetHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil {
writeError(w, http.StatusServiceUnavailable, "client not initialized")
return
}
key := r.URL.Query().Get("key")
if key == "" {
writeError(w, http.StatusBadRequest, "missing 'key'")
return
}
if !g.validateNamespaceParam(r) {
writeError(w, http.StatusForbidden, "namespace mismatch")
return
}
val, err := g.client.Storage().Get(client.WithInternalAuth(r.Context()), key)
if err != nil {
writeError(w, http.StatusNotFound, err.Error())
return
}
w.Header().Set("Content-Type", "application/octet-stream")
w.WriteHeader(http.StatusOK)
_, _ = w.Write(val)
}
func (g *Gateway) storagePutHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil {
writeError(w, http.StatusServiceUnavailable, "client not initialized")
return
}
key := r.URL.Query().Get("key")
if key == "" {
writeError(w, http.StatusBadRequest, "missing 'key'")
return
}
if !g.validateNamespaceParam(r) {
writeError(w, http.StatusForbidden, "namespace mismatch")
return
}
defer r.Body.Close()
b, err := io.ReadAll(r.Body)
if err != nil {
writeError(w, http.StatusBadRequest, "failed to read body")
return
}
if err := g.client.Storage().Put(client.WithInternalAuth(r.Context()), key, b); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusCreated, map[string]any{"status": "ok", "key": key, "size": len(b)})
}
func (g *Gateway) storageDeleteHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil {
writeError(w, http.StatusServiceUnavailable, "client not initialized")
return
}
if !g.validateNamespaceParam(r) {
writeError(w, http.StatusForbidden, "namespace mismatch")
return
}
key := r.URL.Query().Get("key")
if key == "" {
var body struct {
Key string `json:"key"`
Namespace string `json:"namespace"`
}
if err := json.NewDecoder(r.Body).Decode(&body); err == nil {
key = body.Key
}
}
if key == "" {
writeError(w, http.StatusBadRequest, "missing 'key'")
return
}
if err := g.client.Storage().Delete(client.WithInternalAuth(r.Context()), key); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"status": "ok", "key": key})
}
func (g *Gateway) storageListHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil {
writeError(w, http.StatusServiceUnavailable, "client not initialized")
return
}
if !g.validateNamespaceParam(r) {
writeError(w, http.StatusForbidden, "namespace mismatch")
return
}
prefix := r.URL.Query().Get("prefix")
limitStr := r.URL.Query().Get("limit")
limit := 100
if limitStr != "" {
if n, err := strconv.Atoi(limitStr); err == nil && n > 0 {
limit = n
}
}
keys, err := g.client.Storage().List(client.WithInternalAuth(r.Context()), prefix, limit)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"keys": keys})
}
func (g *Gateway) storageExistsHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil {
writeError(w, http.StatusServiceUnavailable, "client not initialized")
return
}
if !g.validateNamespaceParam(r) {
writeError(w, http.StatusForbidden, "namespace mismatch")
return
}
key := r.URL.Query().Get("key")
if key == "" {
writeError(w, http.StatusBadRequest, "missing 'key'")
return
}
exists, err := g.client.Storage().Exists(client.WithInternalAuth(r.Context()), key)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"exists": exists})
}
func (g *Gateway) networkConnectHandler(w http.ResponseWriter, r *http.Request) { func (g *Gateway) networkConnectHandler(w http.ResponseWriter, r *http.Request) {
if g.client == nil { if g.client == nil {
writeError(w, http.StatusServiceUnavailable, "client not initialized") writeError(w, http.StatusServiceUnavailable, "client not initialized")
@ -323,7 +195,7 @@ func (g *Gateway) validateNamespaceParam(r *http.Request) bool {
if qns == "" { if qns == "" {
return true return true
} }
if v := r.Context().Value(storage.CtxKeyNamespaceOverride); v != nil { if v := r.Context().Value(pubsub.CtxKeyNamespaceOverride); v != nil {
if s, ok := v.(string); ok && s != "" { if s, ok := v.(string); ok && s != "" {
return s == qns return s == qns
} }

30
pkg/node/node.go
View File

@ -25,7 +25,6 @@ import (
"git.debros.io/DeBros/network/pkg/config" "git.debros.io/DeBros/network/pkg/config"
"git.debros.io/DeBros/network/pkg/database" "git.debros.io/DeBros/network/pkg/database"
"git.debros.io/DeBros/network/pkg/logging" "git.debros.io/DeBros/network/pkg/logging"
"git.debros.io/DeBros/network/pkg/storage"
) )
// Node represents a network node with RQLite database // Node represents a network node with RQLite database
@ -36,7 +35,6 @@ type Node struct {
rqliteManager *database.RQLiteManager rqliteManager *database.RQLiteManager
rqliteAdapter *database.RQLiteAdapter rqliteAdapter *database.RQLiteAdapter
storageService *storage.Service
// Peer discovery // Peer discovery
discoveryCancel context.CancelFunc discoveryCancel context.CancelFunc
@ -377,24 +375,6 @@ func (n *Node) startLibP2P() error {
return nil return nil
} }
// startStorageService initializes the storage service
func (n *Node) startStorageService() error {
n.logger.ComponentInfo(logging.ComponentStorage, "Starting storage service")
// Create storage service using the RQLite SQL adapter
service, err := storage.NewService(n.rqliteAdapter.GetSQLDB(), n.logger.Logger)
if err != nil {
return err
}
n.storageService = service
// Set up stream handler for storage protocol
n.host.SetStreamHandler("/network/storage/1.0.0", n.storageService.HandleStorageStream)
return nil
}
// loadOrCreateIdentity loads an existing identity or creates a new one // loadOrCreateIdentity loads an existing identity or creates a new one
func (n *Node) loadOrCreateIdentity() (crypto.PrivKey, error) { func (n *Node) loadOrCreateIdentity() (crypto.PrivKey, error) {
identityFile := filepath.Join(n.config.Node.DataDir, "identity.key") identityFile := filepath.Join(n.config.Node.DataDir, "identity.key")
@ -631,11 +611,6 @@ func (n *Node) Stop() error {
// Stop peer discovery // Stop peer discovery
n.stopPeerDiscovery() n.stopPeerDiscovery()
// Stop storage service
if n.storageService != nil {
n.storageService.Close()
}
// Stop LibP2P host // Stop LibP2P host
if n.host != nil { if n.host != nil {
n.host.Close() n.host.Close()
@ -672,11 +647,6 @@ func (n *Node) Start(ctx context.Context) error {
return fmt.Errorf("failed to start LibP2P: %w", err) return fmt.Errorf("failed to start LibP2P: %w", err)
} }
// Start storage service
if err := n.startStorageService(); err != nil {
return fmt.Errorf("failed to start storage service: %w", err)
}
// Get listen addresses for logging // Get listen addresses for logging
var listenAddrs []string var listenAddrs []string
for _, addr := range n.host.Addrs() { for _, addr := range n.host.Addrs() {