mirror of
https://github.com/fiatjaf/khatru.git
synced 2026-04-18 11:26:49 +02:00
adapt to new nip-01.
This commit is contained in:
fiatjaf
133
event.go
133
event.go
@@ -1,133 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/fiatjaf/schnorr"
|
||||
)
|
||||
|
||||
const (
|
||||
KindSetMetadata uint8 = 0
|
||||
KindTextNote uint8 = 1
|
||||
KindRecommendServer uint8 = 2
|
||||
KindContactList uint8 = 3
|
||||
)
|
||||
|
||||
type Event struct {
|
||||
ID string `db:"id" json:"id"` // it's the hash of the serialized event
|
||||
|
||||
PubKey string `db:"pubkey" json:"pubkey"`
|
||||
CreatedAt uint32 `db:"created_at" json:"created_at"`
|
||||
|
||||
Kind uint8 `db:"kind" json:"kind"`
|
||||
|
||||
Tags Tags `db:"tags" json:"tags"`
|
||||
Content string `db:"content" json:"content"`
|
||||
Sig string `db:"sig" json:"sig"`
|
||||
}
|
||||
|
||||
type Tags []Tag
|
||||
|
||||
func (t *Tags) Scan(src interface{}) error {
|
||||
var jtags []byte = make([]byte, 0)
|
||||
|
||||
switch v := src.(type) {
|
||||
case []byte:
|
||||
jtags = v
|
||||
case string:
|
||||
jtags = []byte(v)
|
||||
default:
|
||||
return errors.New("couldn't scan tags, it's not a json string")
|
||||
}
|
||||
|
||||
json.Unmarshal(jtags, &t)
|
||||
return nil
|
||||
}
|
||||
|
||||
type Tag []interface{}
|
||||
|
||||
// Serialize outputs a byte array that can be hashed/signed to identify/authenticate
|
||||
func (evt *Event) Serialize() []byte {
|
||||
// the serialization process is just putting everything into a JSON array
|
||||
// so the order is kept
|
||||
arr := make([]interface{}, 6)
|
||||
|
||||
// version: 0
|
||||
arr[0] = 0
|
||||
|
||||
// pubkey
|
||||
arr[1] = evt.PubKey
|
||||
|
||||
// created_at
|
||||
arr[2] = int64(evt.CreatedAt)
|
||||
|
||||
// kind
|
||||
arr[3] = int64(evt.Kind)
|
||||
|
||||
// tags
|
||||
if evt.Tags != nil {
|
||||
arr[4] = evt.Tags
|
||||
} else {
|
||||
arr[4] = make([]bool, 0)
|
||||
}
|
||||
|
||||
// content
|
||||
arr[5] = evt.Content
|
||||
|
||||
serialized := new(bytes.Buffer)
|
||||
|
||||
enc := json.NewEncoder(serialized)
|
||||
enc.SetEscapeHTML(false)
|
||||
_ = enc.Encode(arr)
|
||||
return serialized.Bytes()[:serialized.Len()-1] // Encode add new line char
|
||||
}
|
||||
|
||||
// CheckSignature checks if the signature is valid for the id
|
||||
// (which is a hash of the serialized event content).
|
||||
// returns an error if the signature itself is invalid.
|
||||
func (evt Event) CheckSignature() (bool, error) {
|
||||
// read and check pubkey
|
||||
pubkeyb, err := hex.DecodeString(evt.PubKey)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if len(pubkeyb) != 32 {
|
||||
return false, fmt.Errorf("pubkey must be 32 bytes, not %d", len(pubkeyb))
|
||||
}
|
||||
|
||||
// check tags
|
||||
for _, tag := range evt.Tags {
|
||||
for _, item := range tag {
|
||||
switch item.(type) {
|
||||
case string, int64, float64, int, bool:
|
||||
// fine
|
||||
default:
|
||||
// not fine
|
||||
return false, fmt.Errorf("tag contains an invalid value %v", item)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
sig, err := hex.DecodeString(evt.Sig)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("signature is invalid hex: %w", err)
|
||||
}
|
||||
if len(sig) != 64 {
|
||||
return false, fmt.Errorf("signature must be 64 bytes, not %d", len(sig))
|
||||
}
|
||||
|
||||
var p [32]byte
|
||||
copy(p[:], pubkeyb)
|
||||
|
||||
var s [64]byte
|
||||
copy(s[:], sig)
|
||||
|
||||
h := sha256.Sum256(evt.Serialize())
|
||||
|
||||
return schnorr.Verify(p, h, s)
|
||||
}
|
||||
9
go.mod
9
go.mod
@@ -3,17 +3,16 @@ module github.com/fiatjaf/nostr-relay
|
||||
go 1.15
|
||||
|
||||
require (
|
||||
github.com/fiatjaf/schnorr v0.2.1-hack
|
||||
github.com/go-sql-driver/mysql v1.5.0 // indirect
|
||||
github.com/fiatjaf/go-nostr v0.0.0-00010101000000-000000000000
|
||||
github.com/gorilla/mux v1.8.0
|
||||
github.com/gorilla/websocket v1.4.2
|
||||
github.com/jmoiron/sqlx v1.2.0
|
||||
github.com/kelseyhightower/envconfig v1.4.0
|
||||
github.com/kr/pretty v0.2.1
|
||||
github.com/lib/pq v1.8.0
|
||||
github.com/mattn/go-sqlite3 v1.14.4
|
||||
github.com/rs/cors v1.7.0
|
||||
github.com/rs/zerolog v1.20.0
|
||||
golang.org/x/time v0.0.0-20201208040808-7e3f01d25324
|
||||
gopkg.in/antage/eventsource.v1 v1.0.0-20150318155416-803f4c5af225
|
||||
google.golang.org/appengine v1.6.7 // indirect
|
||||
)
|
||||
|
||||
replace github.com/fiatjaf/go-nostr => /home/fiatjaf/comp/go-nostr
|
||||
|
||||
22
go.sum
22
go.sum
@@ -15,13 +15,13 @@ github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7
|
||||
github.com/davecgh/go-spew v0.0.0-20171005155431-ecdeabc65495/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/fiatjaf/schnorr v0.2.1-hack h1:6NwQNN5O4+ZUm8KliT+l198vDVH8ovv8AJ8CiL4hvs0=
|
||||
github.com/fiatjaf/schnorr v0.2.1-hack/go.mod h1:6aMsVxPxyO6awpdmNkfkJ8vXqsmUOeGCHp2CdG5LPR0=
|
||||
github.com/fiatjaf/bip340 v1.0.0 h1:mpwbm+0KC9BXB/7/pnac4e4N1TiuppyEVXxtVAXj75k=
|
||||
github.com/fiatjaf/bip340 v1.0.0/go.mod h1:MxAz+5FQUTW4OT2gnCBC6Our486wmqf72ykZIrh7+is=
|
||||
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
|
||||
github.com/go-sql-driver/mysql v1.4.0 h1:7LxgVwFb2hIQtMm87NdgAVfXjnt4OePseqT1tKx+opk=
|
||||
github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
|
||||
github.com/go-sql-driver/mysql v1.5.0 h1:ozyZYNQW3x3HtqT1jira07DN2PArx2v7/mN66gGcHOs=
|
||||
github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
|
||||
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
|
||||
github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
|
||||
github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
|
||||
github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
|
||||
github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
|
||||
@@ -36,11 +36,6 @@ github.com/kelseyhightower/envconfig v1.4.0 h1:Im6hONhd3pLkfDFsbRgu68RDNkGF1r3dv
|
||||
github.com/kelseyhightower/envconfig v1.4.0/go.mod h1:cccZRl6mQpaq41TPp5QxidR+Sa3axMbJDNb//FQX6Gg=
|
||||
github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4=
|
||||
github.com/kkdai/bstream v0.0.0-20181106074824-b3251f7901ec/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4=
|
||||
github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
|
||||
github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
|
||||
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
|
||||
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
||||
github.com/lib/pq v1.0.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
|
||||
github.com/lib/pq v1.8.0 h1:9xohqzkUwzR4Ga4ivdTcawVS89YSDVxXMa3xJX3cGzg=
|
||||
github.com/lib/pq v1.8.0/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
|
||||
@@ -63,6 +58,7 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
|
||||
golang.org/x/net v0.0.0-20180719180050-a680a1efc54d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/net v0.0.0-20190110200230-915654e7eabc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
|
||||
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
@@ -71,12 +67,12 @@ golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5h
|
||||
golang.org/x/sys v0.0.0-20190109145017-48ac38b7c8cb/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/time v0.0.0-20201208040808-7e3f01d25324 h1:Hir2P/De0WpUhtrKGGjvSb2YxUgyZ7EFOSLIcSSpiwE=
|
||||
golang.org/x/time v0.0.0-20201208040808-7e3f01d25324/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
|
||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
golang.org/x/tools v0.0.0-20190828213141-aed303cbaa74/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
gopkg.in/antage/eventsource.v1 v1.0.0-20150318155416-803f4c5af225 h1:xy+AV3uSExoRQc2qWXeZdbhFGwBFK/AmGlrBZEjbvuQ=
|
||||
gopkg.in/antage/eventsource.v1 v1.0.0-20150318155416-803f4c5af225/go.mod h1:SiXNRpUllqhl+GIw2V/BtKI7BUlz+uxov9vBFtXHqh8=
|
||||
google.golang.org/appengine v1.6.7 h1:FZR1q0exgwxzPzp/aF+VccGrSfxfPpkBqjIIEq3ru6c=
|
||||
google.golang.org/appengine v1.6.7/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
|
||||
|
||||
281
handlers.go
281
handlers.go
@@ -2,15 +2,15 @@ package main
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/fiatjaf/go-nostr/event"
|
||||
"github.com/fiatjaf/go-nostr/filter"
|
||||
"github.com/gorilla/websocket"
|
||||
)
|
||||
|
||||
@@ -69,40 +69,66 @@ func handleWebsocket(w http.ResponseWriter, r *http.Request) {
|
||||
continue
|
||||
}
|
||||
|
||||
text := string(message)
|
||||
go func(message []byte) {
|
||||
var err error
|
||||
|
||||
switch {
|
||||
case text == "PING":
|
||||
conn.WriteMessage(websocket.TextMessage, []byte("PONG"))
|
||||
defer func() {
|
||||
if err != nil {
|
||||
conn.WriteJSON([]interface{}{"NOTICE", err.Error()})
|
||||
}
|
||||
}()
|
||||
|
||||
case strings.HasPrefix(text, "{"):
|
||||
// it's a new event
|
||||
err = saveEvent(message)
|
||||
var request []json.RawMessage
|
||||
err = json.Unmarshal(message, &request)
|
||||
if err == nil && len(request) < 2 {
|
||||
err = errors.New("request has less than parameters")
|
||||
return
|
||||
}
|
||||
|
||||
case strings.HasPrefix(text, "sub-key:"):
|
||||
watchPubKey(strings.TrimSpace(text[8:]), conn)
|
||||
var typ string
|
||||
json.Unmarshal(request[0], &typ)
|
||||
|
||||
case strings.HasPrefix(text, "unsub-key:"):
|
||||
unwatchPubKey(strings.TrimSpace(text[10:]), conn)
|
||||
switch typ {
|
||||
case "EVENT":
|
||||
// it's a new event
|
||||
err = saveEvent(request[1])
|
||||
|
||||
case strings.HasPrefix(text, "req-feed:"):
|
||||
err = requestFeed(message[len([]byte("req-feed:")):], conn)
|
||||
case "REQ":
|
||||
var id string
|
||||
json.Unmarshal(request[0], &id)
|
||||
if id == "" {
|
||||
err = errors.New("REQ has no <id>")
|
||||
return
|
||||
}
|
||||
|
||||
case strings.HasPrefix(text, "req-event:"):
|
||||
err = requestEvent(message[len([]byte("req-event:")):], conn)
|
||||
filters := make([]*filter.EventFilter, len(request)-2)
|
||||
for i, filterReq := range request[2:] {
|
||||
err = json.Unmarshal(filterReq, &filters[i])
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
case strings.HasPrefix(text, "req-key:"):
|
||||
err = requestKey(message[len([]byte("req-key:")):], conn)
|
||||
}
|
||||
events, err := queryEvents(filters[i])
|
||||
if err == nil {
|
||||
for _, event := range events {
|
||||
conn.WriteJSON([]interface{}{"EVENT", id, event})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
errj, _ := json.Marshal([]interface{}{
|
||||
"notice",
|
||||
err.Error(),
|
||||
})
|
||||
conn.WriteMessage(websocket.TextMessage, errj)
|
||||
continue
|
||||
}
|
||||
setListener(id, conn, filters)
|
||||
|
||||
case "CLOSE":
|
||||
var id string
|
||||
json.Unmarshal(request[0], &id)
|
||||
if id == "" {
|
||||
err = errors.New("CLOSE has no <id>")
|
||||
return
|
||||
}
|
||||
|
||||
removeListener(id)
|
||||
}
|
||||
}(message)
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -129,7 +155,7 @@ func handleWebsocket(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
func saveEvent(body []byte) error {
|
||||
var evt Event
|
||||
var evt event.Event
|
||||
err := json.Unmarshal(body, &evt)
|
||||
if err != nil {
|
||||
log.Warn().Err(err).Msg("couldn't decode body")
|
||||
@@ -160,16 +186,16 @@ func saveEvent(body []byte) error {
|
||||
|
||||
// react to different kinds of events
|
||||
switch evt.Kind {
|
||||
case KindSetMetadata:
|
||||
case event.KindSetMetadata:
|
||||
// delete past set_metadata events from this user
|
||||
db.Exec(`DELETE FROM event WHERE pubkey = $1 AND kind = 0`, evt.PubKey)
|
||||
case KindTextNote:
|
||||
case event.KindTextNote:
|
||||
// do nothing
|
||||
case KindRecommendServer:
|
||||
case event.KindRecommendServer:
|
||||
// delete past recommend_server events equal to this one
|
||||
db.Exec(`DELETE FROM event WHERE pubkey = $1 AND kind = 2 AND content = $2`,
|
||||
evt.PubKey, evt.Content)
|
||||
case KindContactList:
|
||||
case event.KindContactList:
|
||||
// delete past contact lists from this same pubkey
|
||||
db.Exec(`DELETE FROM event WHERE pubkey = $1 AND kind = 3`, evt.PubKey)
|
||||
}
|
||||
@@ -190,193 +216,6 @@ func saveEvent(body []byte) error {
|
||||
return errors.New("failed to save event")
|
||||
}
|
||||
|
||||
notifyPubKeyEvent(evt.PubKey, &evt)
|
||||
return nil
|
||||
}
|
||||
|
||||
func requestFeed(body []byte, conn *websocket.Conn) error {
|
||||
var data struct {
|
||||
Limit int `json:"limit"`
|
||||
Offset int `json:"offset"`
|
||||
}
|
||||
json.Unmarshal(body, &data)
|
||||
|
||||
if data.Limit <= 0 || data.Limit > 100 {
|
||||
data.Limit = 50
|
||||
}
|
||||
if data.Offset < 0 {
|
||||
data.Offset = 0
|
||||
} else if data.Offset > 500 {
|
||||
return errors.New("offset over 500")
|
||||
}
|
||||
|
||||
keys, ok := backwatchers[conn]
|
||||
if !ok {
|
||||
return errors.New("not subscribed to anything")
|
||||
}
|
||||
|
||||
inkeys := make([]string, 0, len(keys))
|
||||
for _, key := range keys {
|
||||
// to prevent sql attack here we will check if these keys are valid 32byte hex
|
||||
parsed, err := hex.DecodeString(key)
|
||||
if err != nil || len(parsed) != 32 {
|
||||
continue
|
||||
}
|
||||
inkeys = append(inkeys, fmt.Sprintf("'%x'", parsed))
|
||||
}
|
||||
var lastUpdates []Event
|
||||
err := db.Select(&lastUpdates, `
|
||||
SELECT *
|
||||
FROM event
|
||||
WHERE pubkey IN (`+strings.Join(inkeys, ",")+`)
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $1
|
||||
OFFSET $2
|
||||
`, data.Limit, data.Offset)
|
||||
if err != nil && err != sql.ErrNoRows {
|
||||
log.Warn().Err(err).Interface("keys", keys).Msg("failed to fetch events")
|
||||
return errors.New("failed to fetch events")
|
||||
}
|
||||
|
||||
for _, evt := range lastUpdates {
|
||||
jevent, _ := json.Marshal([]interface{}{
|
||||
evt,
|
||||
"p",
|
||||
})
|
||||
conn.WriteMessage(websocket.TextMessage, jevent)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func requestKey(body []byte, conn *websocket.Conn) error {
|
||||
var data struct {
|
||||
Key string `json:"key"`
|
||||
Limit int `json:"limit"`
|
||||
Offset int `json:"offset"`
|
||||
}
|
||||
json.Unmarshal(body, &data)
|
||||
if data.Key == "" {
|
||||
return errors.New("invalid pubkey")
|
||||
}
|
||||
if data.Limit <= 0 || data.Limit > 100 {
|
||||
data.Limit = 30
|
||||
}
|
||||
if data.Offset < 0 {
|
||||
data.Offset = 0
|
||||
} else if data.Offset > 300 {
|
||||
return errors.New("offset over 300")
|
||||
}
|
||||
|
||||
go func() {
|
||||
var metadata Event
|
||||
if err := db.Get(&metadata, `
|
||||
SELECT * FROM event
|
||||
WHERE pubkey = $1 AND kind = 0
|
||||
`, data.Key); err == nil {
|
||||
jevent, _ := json.Marshal([]interface{}{
|
||||
metadata,
|
||||
"r",
|
||||
})
|
||||
conn.WriteMessage(websocket.TextMessage, jevent)
|
||||
} else if err != sql.ErrNoRows {
|
||||
log.Warn().Err(err).
|
||||
Str("key", data.Key).
|
||||
Msg("error fetching metadata from requested user")
|
||||
}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
var lastUpdates []Event
|
||||
if err := db.Select(&lastUpdates, `
|
||||
SELECT * FROM event
|
||||
WHERE pubkey = $1 AND kind != 0
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $2 OFFSET $3
|
||||
`, data.Key, data.Limit, data.Offset); err == nil {
|
||||
for _, evt := range lastUpdates {
|
||||
jevent, _ := json.Marshal([]interface{}{
|
||||
evt,
|
||||
"r",
|
||||
})
|
||||
conn.WriteMessage(websocket.TextMessage, jevent)
|
||||
}
|
||||
} else if err != sql.ErrNoRows {
|
||||
log.Warn().Err(err).
|
||||
Str("key", data.Key).
|
||||
Msg("error fetching updates from requested user")
|
||||
}
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func requestEvent(body []byte, conn *websocket.Conn) error {
|
||||
var data struct {
|
||||
Id string `json:"id"`
|
||||
Limit int `json:"limit"`
|
||||
}
|
||||
json.Unmarshal(body, &data)
|
||||
if data.Id == "" {
|
||||
return errors.New("no id provided")
|
||||
}
|
||||
if data.Limit > 100 || data.Limit <= 0 {
|
||||
data.Limit = 50
|
||||
}
|
||||
|
||||
go func() {
|
||||
// get requested event
|
||||
var evt Event
|
||||
if err := db.Get(&evt, `
|
||||
SELECT * FROM event WHERE id = $1
|
||||
`, data.Id); err == nil {
|
||||
jevent, _ := json.Marshal([]interface{}{
|
||||
evt,
|
||||
"r",
|
||||
})
|
||||
conn.WriteMessage(websocket.TextMessage, jevent)
|
||||
} else if err != sql.ErrNoRows {
|
||||
log.Warn().Err(err).
|
||||
Str("key", data.Id).
|
||||
Msg("error fetching a specific event")
|
||||
}
|
||||
|
||||
for _, tag := range evt.Tags {
|
||||
log.Print(tag)
|
||||
// get referenced event TODO
|
||||
// var ref Event
|
||||
// if err := db.Get(&ref, `
|
||||
// SELECT * FROM event WHERE id = $1
|
||||
// `, evt.Ref); err == nil {
|
||||
// jevent, _ := json.Marshal(ref)
|
||||
// (*es).SendEventMessage(string(jevent), "r", "")
|
||||
// } else if err != sql.ErrNoRows {
|
||||
// log.Warn().Err(err).
|
||||
// Str("key", data.Id).Str("ref", evt.Ref).
|
||||
// Msg("error fetching a referenced event")
|
||||
// }
|
||||
}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
// get events that reference this
|
||||
var related []Event
|
||||
if err := db.Select(&related,
|
||||
relatedEventsQuery,
|
||||
data.Id, data.Limit); err == nil {
|
||||
for _, evt := range related {
|
||||
jevent, _ := json.Marshal([]interface{}{
|
||||
evt,
|
||||
"r",
|
||||
})
|
||||
conn.WriteMessage(websocket.TextMessage, jevent)
|
||||
}
|
||||
} else if err != sql.ErrNoRows {
|
||||
log.Warn().Err(err).
|
||||
Str("key", data.Id).
|
||||
Msg("error fetching events that reference requested event")
|
||||
}
|
||||
}()
|
||||
|
||||
notifyListeners(&evt)
|
||||
return nil
|
||||
}
|
||||
|
||||
112
listener.go
112
listener.go
@@ -1,99 +1,61 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"sync"
|
||||
|
||||
"github.com/fiatjaf/go-nostr/event"
|
||||
"github.com/fiatjaf/go-nostr/filter"
|
||||
"github.com/gorilla/websocket"
|
||||
)
|
||||
|
||||
var watchers = make(map[string][]*websocket.Conn)
|
||||
var backwatchers = make(map[*websocket.Conn][]string)
|
||||
var wlock = sync.Mutex{}
|
||||
type Listener struct {
|
||||
ws *websocket.Conn
|
||||
filters []*filter.EventFilter
|
||||
}
|
||||
|
||||
func watchPubKey(key string, ws *websocket.Conn) {
|
||||
wlock.Lock()
|
||||
defer wlock.Unlock()
|
||||
var listeners = make(map[string]*Listener)
|
||||
var listenersMutex = sync.Mutex{}
|
||||
|
||||
currentKeys, _ := backwatchers[ws]
|
||||
backwatchers[ws] = append(currentKeys, key)
|
||||
func setListener(id string, conn *websocket.Conn, filters []*filter.EventFilter) {
|
||||
listenersMutex.Lock()
|
||||
defer func() {
|
||||
listenersMutex.Unlock()
|
||||
}()
|
||||
|
||||
if wss, ok := watchers[key]; ok {
|
||||
watchers[key] = append(wss, ws)
|
||||
} else {
|
||||
watchers[key] = []*websocket.Conn{ws}
|
||||
listeners[id] = &Listener{
|
||||
ws: conn,
|
||||
filters: filters,
|
||||
}
|
||||
}
|
||||
|
||||
func unwatchPubKey(excludedKey string, ws *websocket.Conn) {
|
||||
wlock.Lock()
|
||||
defer wlock.Unlock()
|
||||
func removeListener(id string) {
|
||||
listenersMutex.Lock()
|
||||
defer func() {
|
||||
listenersMutex.Unlock()
|
||||
}()
|
||||
|
||||
if wss, ok := watchers[excludedKey]; ok {
|
||||
newWss := make([]*websocket.Conn, len(wss)-1)
|
||||
delete(listeners, id)
|
||||
}
|
||||
|
||||
var i = 0
|
||||
for _, existingWs := range wss {
|
||||
if existingWs == ws {
|
||||
continue
|
||||
func notifyListeners(event *event.Event) {
|
||||
listenersMutex.Lock()
|
||||
defer func() {
|
||||
listenersMutex.Unlock()
|
||||
}()
|
||||
|
||||
for id, listener := range listeners {
|
||||
match := false
|
||||
for _, filter := range listener.filters {
|
||||
if filter.Matches(event) {
|
||||
match = true
|
||||
break
|
||||
}
|
||||
if i == len(wss) {
|
||||
// if we reach this point it is because the key we were
|
||||
// excluding wasn't here in the first place
|
||||
return
|
||||
}
|
||||
newWss[i] = existingWs
|
||||
i++
|
||||
}
|
||||
|
||||
watchers[excludedKey] = newWss
|
||||
}
|
||||
|
||||
currentKeys, _ := backwatchers[ws]
|
||||
newKeys := make([]string, 0, len(currentKeys))
|
||||
for _, currentKey := range currentKeys {
|
||||
if excludedKey == currentKey {
|
||||
if !match {
|
||||
continue
|
||||
}
|
||||
newKeys = append(newKeys, currentKey)
|
||||
}
|
||||
|
||||
backwatchers[ws] = newKeys
|
||||
}
|
||||
|
||||
func removeFromWatchers(es *websocket.Conn) {
|
||||
wlock.Lock()
|
||||
defer wlock.Unlock()
|
||||
|
||||
for _, key := range backwatchers[es] {
|
||||
if arr, ok := watchers[key]; ok {
|
||||
newarr := make([]*websocket.Conn, len(arr)-1)
|
||||
i := 0
|
||||
for _, oldes := range arr {
|
||||
if oldes == es {
|
||||
continue
|
||||
}
|
||||
newarr[i] = oldes
|
||||
i++
|
||||
}
|
||||
watchers[key] = newarr
|
||||
}
|
||||
}
|
||||
delete(backwatchers, es)
|
||||
}
|
||||
|
||||
func notifyPubKeyEvent(key string, evt *Event) {
|
||||
wlock.Lock()
|
||||
arr, ok := watchers[key]
|
||||
wlock.Unlock()
|
||||
|
||||
if ok {
|
||||
for _, conn := range arr {
|
||||
jevent, _ := json.Marshal([]interface{}{
|
||||
evt,
|
||||
"n",
|
||||
})
|
||||
conn.WriteMessage(websocket.TextMessage, jevent)
|
||||
}
|
||||
listener.ws.WriteJSON([]interface{}{"EVENT", id, event})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -31,8 +31,4 @@ CREATE INDEX pubkeytimeidx ON event (pubkey, created_at);
|
||||
return db, nil
|
||||
}
|
||||
|
||||
const relatedEventsQuery = `
|
||||
SELECT * FROM event
|
||||
WHERE tags @@ '$[*][1] == "' || $1 || '"'
|
||||
LIMIT $2
|
||||
`
|
||||
const relatedEventsCondition = `tags @@ '$[*][1] == "' || ? || '"'`
|
||||
|
||||
71
query.go
Normal file
71
query.go
Normal file
@@ -0,0 +1,71 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/fiatjaf/go-nostr/event"
|
||||
"github.com/fiatjaf/go-nostr/filter"
|
||||
)
|
||||
|
||||
func queryEvents(filter *filter.EventFilter) (events []event.Event, err error) {
|
||||
var conditions []string
|
||||
var params []interface{}
|
||||
|
||||
if filter.ID != "" {
|
||||
conditions = append(conditions, "id = ?")
|
||||
params = append(params, filter.ID)
|
||||
}
|
||||
|
||||
if filter.Author != "" {
|
||||
conditions = append(conditions, "pubkey = ?")
|
||||
params = append(params, filter.Author)
|
||||
}
|
||||
|
||||
if filter.Authors != nil {
|
||||
inkeys := make([]string, 0, len(filter.Authors))
|
||||
for _, key := range filter.Authors {
|
||||
// to prevent sql attack here we will check if
|
||||
// these keys are valid 32byte hex
|
||||
parsed, err := hex.DecodeString(key)
|
||||
if err != nil || len(parsed) != 32 {
|
||||
continue
|
||||
}
|
||||
inkeys = append(inkeys, fmt.Sprintf("'%x'", parsed))
|
||||
}
|
||||
conditions = append(conditions, `pubkey IN (`+strings.Join(inkeys, ",")+`)`)
|
||||
}
|
||||
|
||||
if filter.TagEvent != "" {
|
||||
conditions = append(conditions, relatedEventsCondition)
|
||||
params = append(params, filter.TagEvent)
|
||||
}
|
||||
|
||||
if filter.TagProfile != "" {
|
||||
conditions = append(conditions, relatedEventsCondition)
|
||||
params = append(params, filter.TagProfile)
|
||||
}
|
||||
|
||||
if filter.Since != 0 {
|
||||
conditions = append(conditions, "created_at > ?")
|
||||
params = append(params, filter.Since)
|
||||
}
|
||||
|
||||
if len(conditions) == 0 {
|
||||
// fallback
|
||||
conditions = append(conditions, "true")
|
||||
}
|
||||
|
||||
err = db.Select(&events, "SELECT * FROM event WHERE "+
|
||||
strings.Join(conditions, " AND ")+
|
||||
"ORDER BY created_at LIMIT 100", params...,
|
||||
)
|
||||
if err != nil && err != sql.ErrNoRows {
|
||||
log.Warn().Err(err).Interface("filter", filter).Msg("failed to fetch events")
|
||||
err = fmt.Errorf("failed to fetch events: %w", err)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
Reference in New Issue
Block a user