From ecef7b5fb8d239ece387d3024ecd21a08c309f24 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Mon, 13 Dec 2021 04:07:38 +0100 Subject: [PATCH] Webapi: Live implementation of search and explore. Removed dummy code. --- webapi/API.go | 87 ++++++++------ webapi/Download Temp.go | 2 +- webapi/Search Dispatch.go | 65 ++++++++++ webapi/Search Temp.go | 246 -------------------------------------- webapi/Search.go | 10 +- webapi/Shared Recent.go | 75 ++++++++++++ 6 files changed, 194 insertions(+), 291 deletions(-) create mode 100644 webapi/Search Dispatch.go delete mode 100644 webapi/Search Temp.go create mode 100644 webapi/Shared Recent.go diff --git a/webapi/API.go b/webapi/API.go index 2578625..dc32676 100644 --- a/webapi/API.go +++ b/webapi/API.go @@ -19,8 +19,12 @@ import ( "github.com/gorilla/websocket" ) -// Router can be used to register additional API functions -var Router *mux.Router +type WebapiInstance struct { + backend *core.Backend + + // Router can be used to register additional API functions + Router *mux.Router +} // WSUpgrader is used for websocket functionality. It allows all requests. var WSUpgrader = websocket.Upgrader{ @@ -35,53 +39,58 @@ var WSUpgrader = websocket.Upgrader{ // Start starts the API. ListenAddresses is a list of IP:Ports. // The certificate file and key are only used if SSL is enabled. The read and write timeout may be 0 for no timeout. // The API key may be uuid.Nil to disable it although this is not recommended for security reasons. -func Start(ListenAddresses []string, UseSSL bool, CertificateFile, CertificateKey string, TimeoutRead, TimeoutWrite time.Duration, APIKey uuid.UUID) { +func Start(Backend *core.Backend, ListenAddresses []string, UseSSL bool, CertificateFile, CertificateKey string, TimeoutRead, TimeoutWrite time.Duration, APIKey uuid.UUID) (api *WebapiInstance) { if len(ListenAddresses) == 0 { - return + return nil } - Router = mux.NewRouter() + api = &WebapiInstance{ + backend: Backend, + Router: mux.NewRouter(), + } if APIKey != uuid.Nil { - Router.Use(authenticateMiddleware(APIKey)) + api.Router.Use(authenticateMiddleware(APIKey)) } - Router.HandleFunc("/test", apiTest).Methods("GET") - Router.HandleFunc("/status", apiStatus).Methods("GET") - Router.HandleFunc("/account/info", apiAccountInfo).Methods("GET") - Router.HandleFunc("/account/delete", apiAccountDelete).Methods("GET") - Router.HandleFunc("/blockchain/header", apiBlockchainHeaderFunc).Methods("GET") - Router.HandleFunc("/blockchain/append", apiBlockchainAppend).Methods("POST") - Router.HandleFunc("/blockchain/read", apiBlockchainRead).Methods("GET") - Router.HandleFunc("/blockchain/file/add", apiBlockchainFileAdd).Methods("POST") - Router.HandleFunc("/blockchain/file/list", apiBlockchainFileList).Methods("GET") - Router.HandleFunc("/blockchain/file/delete", apiBlockchainFileDelete).Methods("POST") - Router.HandleFunc("/blockchain/file/update", apiBlockchainFileUpdate).Methods("POST") - Router.HandleFunc("/profile/list", apiProfileList).Methods("GET") - Router.HandleFunc("/profile/read", apiProfileRead).Methods("GET") - Router.HandleFunc("/profile/write", apiProfileWrite).Methods("POST") - Router.HandleFunc("/profile/delete", apiProfileDelete).Methods("POST") - Router.HandleFunc("/search", apiSearch).Methods("POST") - Router.HandleFunc("/search/result", apiSearchResult).Methods("GET") - Router.HandleFunc("/search/result/ws", apiSearchResultStream).Methods("GET") - Router.HandleFunc("/search/statistic", apiSearchStatistic).Methods("GET") - Router.HandleFunc("/search/terminate", apiSearchTerminate).Methods("GET") - Router.HandleFunc("/explore", apiExplore).Methods("GET") - Router.HandleFunc("/file/format", apiFileFormat).Methods("GET") - Router.HandleFunc("/download/start", apiDownloadStart).Methods("GET") - Router.HandleFunc("/download/status", apiDownloadStatus).Methods("GET") - Router.HandleFunc("/download/action", apiDownloadAction).Methods("GET") - Router.HandleFunc("/warehouse/create", apiWarehouseCreateFile).Methods("POST") - Router.HandleFunc("/warehouse/create/path", apiWarehouseCreateFilePath).Methods("GET") - Router.HandleFunc("/warehouse/read", apiWarehouseReadFile).Methods("GET") - Router.HandleFunc("/warehouse/read/path", apiWarehouseReadFilePath).Methods("GET") - Router.HandleFunc("/warehouse/delete", apiWarehouseDeleteFile).Methods("GET") - Router.HandleFunc("/file/read", apiFileRead).Methods("GET") - Router.HandleFunc("/file/view", apiFileView).Methods("GET") + api.Router.HandleFunc("/test", apiTest).Methods("GET") + api.Router.HandleFunc("/status", apiStatus).Methods("GET") + api.Router.HandleFunc("/account/info", apiAccountInfo).Methods("GET") + api.Router.HandleFunc("/account/delete", apiAccountDelete).Methods("GET") + api.Router.HandleFunc("/blockchain/header", apiBlockchainHeaderFunc).Methods("GET") + api.Router.HandleFunc("/blockchain/append", apiBlockchainAppend).Methods("POST") + api.Router.HandleFunc("/blockchain/read", apiBlockchainRead).Methods("GET") + api.Router.HandleFunc("/blockchain/file/add", apiBlockchainFileAdd).Methods("POST") + api.Router.HandleFunc("/blockchain/file/list", apiBlockchainFileList).Methods("GET") + api.Router.HandleFunc("/blockchain/file/delete", apiBlockchainFileDelete).Methods("POST") + api.Router.HandleFunc("/blockchain/file/update", apiBlockchainFileUpdate).Methods("POST") + api.Router.HandleFunc("/profile/list", apiProfileList).Methods("GET") + api.Router.HandleFunc("/profile/read", apiProfileRead).Methods("GET") + api.Router.HandleFunc("/profile/write", apiProfileWrite).Methods("POST") + api.Router.HandleFunc("/profile/delete", apiProfileDelete).Methods("POST") + api.Router.HandleFunc("/search", api.apiSearch).Methods("POST") + api.Router.HandleFunc("/search/result", apiSearchResult).Methods("GET") + api.Router.HandleFunc("/search/result/ws", apiSearchResultStream).Methods("GET") + api.Router.HandleFunc("/search/statistic", apiSearchStatistic).Methods("GET") + api.Router.HandleFunc("/search/terminate", apiSearchTerminate).Methods("GET") + api.Router.HandleFunc("/explore", api.apiExplore).Methods("GET") + api.Router.HandleFunc("/file/format", apiFileFormat).Methods("GET") + api.Router.HandleFunc("/download/start", apiDownloadStart).Methods("GET") + api.Router.HandleFunc("/download/status", apiDownloadStatus).Methods("GET") + api.Router.HandleFunc("/download/action", apiDownloadAction).Methods("GET") + api.Router.HandleFunc("/warehouse/create", apiWarehouseCreateFile).Methods("POST") + api.Router.HandleFunc("/warehouse/create/path", apiWarehouseCreateFilePath).Methods("GET") + api.Router.HandleFunc("/warehouse/read", apiWarehouseReadFile).Methods("GET") + api.Router.HandleFunc("/warehouse/read/path", apiWarehouseReadFilePath).Methods("GET") + api.Router.HandleFunc("/warehouse/delete", apiWarehouseDeleteFile).Methods("GET") + api.Router.HandleFunc("/file/read", apiFileRead).Methods("GET") + api.Router.HandleFunc("/file/view", apiFileView).Methods("GET") for _, listen := range ListenAddresses { - go startWebAPI(listen, UseSSL, CertificateFile, CertificateKey, Router, "API", TimeoutRead, TimeoutWrite) + go startWebAPI(listen, UseSSL, CertificateFile, CertificateKey, api.Router, "API", TimeoutRead, TimeoutWrite) } + + return api } // startWebAPI starts a web-server with given parameters and logs the status. If may block forever and only returns if there is an error. diff --git a/webapi/Download Temp.go b/webapi/Download Temp.go index f7c2b66..07d2b00 100644 --- a/webapi/Download Temp.go +++ b/webapi/Download Temp.go @@ -19,7 +19,7 @@ func (info *downloadInfo) Start() { time.Sleep(time.Second * time.Duration(rand.Intn(5))) // request metadata - info.file = blockRecordFileToAPI(createTestResult(-1)) + //info.file = blockRecordFileToAPI(createTestResult(-1)) // join swarm diff --git a/webapi/Search Dispatch.go b/webapi/Search Dispatch.go new file mode 100644 index 0000000..9c4f259 --- /dev/null +++ b/webapi/Search Dispatch.go @@ -0,0 +1,65 @@ +/* +File Name: Search Dispatch.go +Copyright: 2021 Peernet Foundation s.r.o. +Author: Peter Kleissner +*/ +package webapi + +import ( + "bytes" + "time" + + "github.com/PeernetOfficial/core" +) + +func (api *WebapiInstance) dispatchSearch(input SearchRequest) (job *SearchJob) { + Timeout := input.Parse() + Filter := input.ToSearchFilter() + + // create the search job + job = CreateSearchJob(Timeout, input.MaxResults, Filter) + + // todo: create actual search clients! + + go job.localSearch(api.backend, input.Term) + + job.RemoveDefer(job.timeout + time.Minute*10) + + return job +} + +func (job *SearchJob) localSearch(backend *core.Backend, term string) { + if backend.SearchIndex == nil || backend.GlobalBlockchainCache == nil { + return + } + + results := backend.SearchIndex.Search(term) + + job.ResultSync.Lock() + +resultLoop: + for _, result := range results { + file, _, found, err := backend.GlobalBlockchainCache.ReadFile(result.PublicKey, result.BlockchainVersion, result.BlockNumber, result.FileID) + if err != nil || !found { + continue + } + + // new result + newFile := blockRecordFileToAPI(file) + + // Deduplicate based on file hash from the same peer. + for n := range job.AllFiles { + if bytes.Equal(job.AllFiles[n].Hash, newFile.Hash) && bytes.Equal(job.AllFiles[n].NodeID, newFile.NodeID) { + continue resultLoop + } + } + + job.Files = append(job.Files, &newFile) + job.AllFiles = append(job.AllFiles, &newFile) + job.requireSort = true + job.statsAdd(&newFile) + } + + job.ResultSync.Unlock() + job.Terminate() +} diff --git a/webapi/Search Temp.go b/webapi/Search Temp.go deleted file mode 100644 index 4a14fbe..0000000 --- a/webapi/Search Temp.go +++ /dev/null @@ -1,246 +0,0 @@ -/* -File Name: Search Temp.go -Copyright: 2021 Peernet Foundation s.r.o. -Author: Peter Kleissner - -Temporary search code to provide dummy results for testing. To be replaced! -*/ - -package webapi - -import ( - "encoding/hex" - "fmt" - "math/rand" - "time" - - "github.com/PeernetOfficial/core" - "github.com/PeernetOfficial/core/blockchain" - "github.com/google/uuid" -) - -func dispatchSearch(input SearchRequest) (job *SearchJob) { - Timeout := input.Parse() - Filter := input.ToSearchFilter() - - // create the search job - job = CreateSearchJob(Timeout, input.MaxResults, Filter) - - // Create test data - // * Between 0-100 results - // * Delay of first result is 0-5 seconds - // * File name, type, format, ID and hash are randomized - go func(job *SearchJob) { - rand.Seed(time.Now().UnixNano()) - waitTime := time.Duration(rand.Intn(5)) * time.Second - countResults := rand.Intn(100) - - time.Sleep(waitTime) - job.ResultSync.Lock() - - for n := 0; n < countResults; n++ { - newFile := blockRecordFileToAPI(createTestResult(-1)) - - // TODO: Move to channel! - job.Files = append(job.Files, &newFile) - job.AllFiles = append(job.AllFiles, &newFile) - job.requireSort = true - job.statsAdd(&newFile) - } - - job.ResultSync.Unlock() - job.Terminate() - - }(job) - - job.RemoveDefer(job.timeout + time.Minute*10) - - return job -} - -// createTestResult creates a test file. fileType = -1 for any. -func createTestResult(fileType int) (file blockchain.BlockRecordFile) { - randomData := make([]byte, 10*1024) - rand.Read(randomData) - - file.Hash = core.Data2Hash(randomData) - file.Format = uint16(rand.Intn(core.FormatCSV)) - file.ID = uuid.New() - file.Size = uint64(len(randomData)) - - file.NodeID = make([]byte, 32) // node ID = blake3 hash of peer ID - rand.Read(file.NodeID) - - if fileType == -1 { - switch file.Format { - case core.FormatCSV, core.FormatEmail, core.FormatText, core.FormatHTML: - file.Type = core.TypeText - - case core.FormatDatabase: - file.Type = core.TypeBinary - - case core.FormatCompressed: - file.Type = core.TypeCompressed - - case core.FormatContainer: - file.Type = core.TypeContainer - - case core.FormatEbook: - file.Type = core.TypeEbook - - case core.FormatVideo: - file.Type = core.TypeVideo - - case core.FormatAudio: - file.Type = core.TypeAudio - - case core.FormatPicture: - file.Type = core.TypePicture - - case core.FormatPowerpoint, core.FormatExcel, core.FormatWord, core.FormatPDF: - file.Type = core.TypeDocument - - case core.FormatFolder: - file.Type = core.TypeFolder - - default: - file.Type = core.TypeBinary - - } - } else { - if fileType == -2 { - // Binary, Compressed, Container, Executable - otherList := []int{core.TypeBinary, core.TypeCompressed, core.TypeContainer, core.TypeExecutable} - fileType = otherList[rand.Intn(len(otherList))] - } - - file.Type = uint8(fileType) - switch file.Type { - case core.TypeBinary: - file.Format = core.FormatBinary - - case core.TypeText: - file.Format = core.FormatText - - case core.TypePicture: - file.Format = core.FormatPicture - - case core.TypeVideo: - file.Format = core.FormatVideo - - case core.TypeAudio: - file.Format = core.FormatAudio - - case core.TypeDocument: - file.Format = core.FormatPDF - - case core.TypeExecutable: - file.Format = core.FormatExecutable - - case core.TypeContainer: - file.Format = core.FormatContainer - - case core.TypeCompressed: - file.Format = core.FormatCompressed - - case core.TypeFolder: - file.Format = core.FormatFolder - - case core.TypeEbook: - file.Format = core.FormatEbook - - default: - file.Format = core.FormatBinary - } - } - - var extension string - switch file.Type { - case core.TypeBinary: - extension = "bin" - - case core.TypeText: - extension = "txt" - - case core.TypePicture: - extension = "jpg" - - case core.TypeVideo: - extension = "mp4" - - case core.TypeAudio: - extension = "mp3" - - case core.TypeDocument: - extension = "pdf" - - case core.TypeExecutable: - extension = "exe" - - case core.TypeContainer: - extension = "zip" - - case core.TypeCompressed: - extension = "gz" - - case core.TypeFolder: - extension = "" - - case core.TypeEbook: - extension = "" - - default: - extension = ".bin" - } - - file.Tags = append(file.Tags, blockchain.TagFromText(blockchain.TagName, tempFileName("", extension))) - //file.Tags = append(file.Tags, blockchain.TagFromText(blockchain.TagFolder, "not set")) - file.Tags = append(file.Tags, blockchain.TagFromDate(blockchain.TagDateShared, time.Now().UTC())) - - sharedByCount := uint64(rand.Intn(10)) - file.Tags = append(file.Tags, blockchain.TagFromNumber(blockchain.TagSharedByCount, sharedByCount)) - - if sharedByCount > 0 { - var sharedByGeoIP string - for n := uint64(0); n < sharedByCount; n++ { - latitude, longitude := randomGeoIP() - if n > 0 { - sharedByGeoIP += "\n" - } - - sharedByGeoIP += fmt.Sprintf("%.4f", latitude) + "," + fmt.Sprintf("%.4f", longitude) - } - - file.Tags = append(file.Tags, blockchain.TagFromText(blockchain.TagSharedByGeoIP, sharedByGeoIP)) - } - - return -} - -// tempFileName generates a temporary filename for use in testing -func tempFileName(prefix, suffix string) string { - randBytes := make([]byte, 16) - rand.Read(randBytes) - return prefix + hex.EncodeToString(randBytes) + "." + suffix -} - -// randomGeoIP generates random geo IP coordinates -func randomGeoIP() (latitude, longitude float32) { - // Latitude generates latitude (from -90.0 to 90.0) - latitude = rand.Float32()*180 - 90 - - // Longitude generates longitude (from -180 to 180) - longitude = rand.Float32()*360 - 180 - - return -} - -// queryRecentShared returns recently shared files on the network. fileType = -1 for any. -func queryRecentShared(fileType, limit int) (files []*blockchain.BlockRecordFile) { - for n := 0; n < limit; n++ { - newFile := createTestResult(fileType) - files = append(files, &newFile) - } - - return -} diff --git a/webapi/Search.go b/webapi/Search.go index 7b05987..61a55ab 100644 --- a/webapi/Search.go +++ b/webapi/Search.go @@ -82,7 +82,7 @@ Request: POST /search with JSON SearchRequest Result: 200 on success with JSON SearchRequestResponse 400 on invalid JSON */ -func apiSearch(w http.ResponseWriter, r *http.Request) { +func (api *WebapiInstance) apiSearch(w http.ResponseWriter, r *http.Request) { var input SearchRequest if err := DecodeJSON(w, r, &input); err != nil { @@ -104,7 +104,7 @@ func apiSearch(w http.ResponseWriter, r *http.Request) { } } - job := dispatchSearch(input) + job := api.dispatchSearch(input) EncodeJSON(w, r, SearchRequestResponse{Status: 0, ID: job.id}) } @@ -332,7 +332,7 @@ Special type -2 = Binary, Compressed, Container, Executable. This special type i Request: GET /explore?limit=[max records]&type=[file type] Result: 200 with JSON structure SearchResult. Check the field status. */ -func apiExplore(w http.ResponseWriter, r *http.Request) { +func (api *WebapiInstance) apiExplore(w http.ResponseWriter, r *http.Request) { r.ParseForm() limit, err := strconv.Atoi(r.Form.Get("limit")) if err != nil { @@ -343,14 +343,14 @@ func apiExplore(w http.ResponseWriter, r *http.Request) { fileType = -1 } - resultFiles := queryRecentShared(fileType, limit) + resultFiles := queryRecentShared(api.backend, fileType, uint64(limit*20/100), uint64(limit)) var result SearchResult result.Files = []apiFile{} // loop over results for n := range resultFiles { - result.Files = append(result.Files, blockRecordFileToAPI(*resultFiles[n])) + result.Files = append(result.Files, blockRecordFileToAPI(resultFiles[n])) } if len(result.Files) == 0 { diff --git a/webapi/Shared Recent.go b/webapi/Shared Recent.go new file mode 100644 index 0000000..c7aaa33 --- /dev/null +++ b/webapi/Shared Recent.go @@ -0,0 +1,75 @@ +/* +File Name: Shared Recent.go +Copyright: 2021 Peernet Foundation s.r.o. +Author: Peter Kleissner +*/ +package webapi + +import ( + "github.com/PeernetOfficial/core" + "github.com/PeernetOfficial/core/blockchain" +) + +// queryRecentShared returns recently shared files on the network from random peers until the limit is reached. +func queryRecentShared(backend *core.Backend, fileType int, limitPeer, limitTotal uint64) (files []blockchain.BlockRecordFile) { + if limitPeer == 0 { + limitPeer = 1 + } + + // Use the peer list to know about active peers. Random order! + peerList := core.PeerlistGet() + + // Files from peers exceeding the limit. It is used if from all peers the total limit is not reached. + var filesSeconday []blockchain.BlockRecordFile + + for _, peer := range peerList { + if peer.BlockchainHeight == 0 { + continue + } + + var filesFromPeer uint64 + + // decode blocks from top down + blockLoop: + for blockN := peer.BlockchainHeight - 1; blockN > 0; blockN-- { + blockDecoded, _, found, _ := backend.GlobalBlockchainCache.ReadBlock(peer.PublicKey, peer.BlockchainVersion, blockN) + if !found { + continue + } + + for _, record := range blockDecoded.RecordsDecoded { + if file, ok := record.(blockchain.BlockRecordFile); ok && isFileTypeMatchBlock(&file, fileType) { + // found a new file! append. + if filesFromPeer < limitPeer { + files = append(files, file) + + if uint64(len(files)) >= limitTotal { + return + } + + filesFromPeer++ + } else if uint64(len(filesSeconday)) < limitTotal-uint64(len(files)) { + filesSeconday = append(filesSeconday, file) + } else { + break blockLoop + } + } + } + } + } + + files = append(files, filesSeconday...) + + return +} + +// isFileTypeMatchBlock checks if the file type matches. -1 = accept any. -2 = core.TypeBinary, core.TypeCompressed, core.TypeContainer, core.TypeExecutable. +func isFileTypeMatchBlock(file *blockchain.BlockRecordFile, fileType int) bool { + if fileType == -1 { + return true + } else if fileType == -2 { + return file.Type == core.TypeBinary || file.Type == core.TypeCompressed || file.Type == core.TypeContainer || file.Type == core.TypeExecutable + } + + return file.Type == uint8(fileType) +}