From c053c617c2a987b9fbff74c0599ef754103144e4 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Tue, 14 Dec 2021 15:45:03 +0100 Subject: [PATCH] Add users blockchain data to the search index and provide it in search results. Updated webapi search and download. --- Blockchain Cache Global.go | 19 +++++++++++++++++-- Blockchain User.go | 35 +++++++++++++++++++++++++++++++++++ Peernet.go | 5 +++-- blockchain/Blockchain.go | 11 +++++++++++ webapi/Download Transfer.go | 33 +++++++++++++++++++++++++++++++++ webapi/Search Dispatch.go | 5 ++++- 6 files changed, 103 insertions(+), 5 deletions(-) diff --git a/Blockchain Cache Global.go b/Blockchain Cache Global.go index dd01fd6..0ed3e08 100644 --- a/Blockchain Cache Global.go +++ b/Blockchain Cache Global.go @@ -163,8 +163,23 @@ func (cache *BlockchainCache) ReadFile(PublicKey *btcec.PublicKey, Version, Bloc // ReadBlock reads a block and decodes the records. func (cache *BlockchainCache) ReadBlock(PublicKey *btcec.PublicKey, Version, BlockNumber uint64) (decoded *blockchain.BlockDecoded, raw []byte, found bool, err error) { - if raw, found = cache.store.ReadBlock(PublicKey, Version, BlockNumber); !found { - return nil, nil, false, nil + // requesting a block from the user's blockchain? + if PublicKey.IsEqual(peerPublicKey) { + _, _, version := UserBlockchain.Header() + if Version != version { + return nil, nil, false, nil + } + + var status int + raw, status, err = UserBlockchain.GetBlockRaw(BlockNumber) + if err != nil || status != blockchain.StatusOK { + return nil, raw, false, err + } + } else { + // read from the cache + if raw, found = cache.store.ReadBlock(PublicKey, Version, BlockNumber); !found { + return nil, nil, false, nil + } } // decode the entire block diff --git a/Blockchain User.go b/Blockchain User.go index 02eebaa..5e51d91 100644 --- a/Blockchain User.go +++ b/Blockchain User.go @@ -26,3 +26,38 @@ func initUserBlockchain() { os.Exit(ExitBlockchainCorrupt) } } + +// Index the user's blockchain each time there is an update. +func (backend *Backend) userBlockchainUpdateSearchIndex() { + UserBlockchain.BlockchainUpdate = func(blockchainU *blockchain.Blockchain, oldHeight, oldVersion, newHeight, newVersion uint64) { + + if newVersion != oldVersion || newHeight < oldHeight { + // invalidate search index data for the user's blockchain + backend.SearchIndex.UnindexBlockchain(peerPublicKey) + + // reindex everything + for blockN := uint64(0); blockN < newHeight; blockN++ { + raw, status, err := blockchainU.GetBlockRaw(blockN) + if err != nil || status != blockchain.StatusOK { + continue + } + + backend.SearchIndex.IndexNewBlock(peerPublicKey, newVersion, blockN, raw) + } + + return + } + + if newVersion == oldVersion && newHeight > oldHeight { + // index the new blocks + for blockN := oldHeight; blockN < newHeight; blockN++ { + raw, status, err := blockchainU.GetBlockRaw(blockN) + if err != nil || status != blockchain.StatusOK { + continue + } + + backend.SearchIndex.IndexNewBlock(peerPublicKey, newVersion, blockN, raw) + } + } + } +} diff --git a/Peernet.go b/Peernet.go index 1866b4a..da53eae 100644 --- a/Peernet.go +++ b/Peernet.go @@ -37,10 +37,11 @@ func Init(UserAgent string) (backend *Backend) { var err error backend.GlobalBlockchainCache = initBlockchainCache(config.BlockchainGlobal, config.CacheMaxBlockSize, config.CacheMaxBlockCount, config.LimitTotalRecords) - backend.SearchIndex, err = search.InitSearchIndexStore(config.SearchIndex) - if err != nil { + if backend.SearchIndex, err = search.InitSearchIndexStore(config.SearchIndex); err != nil { Filters.LogError("Init", "search index '%s' init: %s", config.SearchIndex, err.Error()) + } else { + backend.userBlockchainUpdateSearchIndex() } return backend diff --git a/blockchain/Blockchain.go b/blockchain/Blockchain.go index 89cd164..bc47ecf 100644 --- a/blockchain/Blockchain.go +++ b/blockchain/Blockchain.go @@ -47,6 +47,9 @@ type Blockchain struct { path string // Path of the blockchain on disk. Depends on key-value store whether a filename or folder. database store.Store // The database storing the blockchain. sync.Mutex // synchronized access to the header + + // callback + BlockchainUpdate func(blockchain *Blockchain, oldHeight, oldVersion, newHeight, newVersion uint64) } // Init initializes the given blockchain. It creates the blockchain file if it does not exist already. @@ -109,6 +112,9 @@ func (blockchain *Blockchain) headerRead() (found bool, err error) { // headerWrite writes the header to the blockchain and signs it. func (blockchain *Blockchain) headerWrite(height, version uint64) (err error) { + oldHeight := blockchain.height + oldVersion := blockchain.version + blockchain.height = height blockchain.version = version @@ -129,6 +135,11 @@ func (blockchain *Blockchain) headerWrite(height, version uint64) (err error) { err = blockchain.database.Set([]byte(keyHeader), buffer[:]) + // call the callback, if any + if blockchain.BlockchainUpdate != nil { + blockchain.BlockchainUpdate(blockchain, oldHeight, oldVersion, blockchain.height, blockchain.version) + } + return err } diff --git a/webapi/Download Transfer.go b/webapi/Download Transfer.go index 2312c87..5e848f7 100644 --- a/webapi/Download Transfer.go +++ b/webapi/Download Transfer.go @@ -9,14 +9,22 @@ Temporary download code to provide dummy results for testing. To be replaced! package webapi import ( + "bytes" "os" "time" "github.com/PeernetOfficial/core" + "github.com/PeernetOfficial/core/warehouse" ) // Starts the download. func (info *downloadInfo) Start() { + // current user? + if bytes.Equal(info.nodeID, core.SelfNodeID()) { + info.DownloadSelf() + return + } + for n := 0; n < 3 && info.peer == nil; n++ { _, info.peer, _ = core.FindNode(info.nodeID, time.Second*5) @@ -168,3 +176,28 @@ func (info *downloadInfo) storeDownloadData(data []byte, offset uint64) (status return DownloadResponseSuccess } + +func (info *downloadInfo) DownloadSelf() { + // Check if the file is available in the local warehouse. + _, fileInfo, status, _ := core.UserWarehouse.FileExists(info.hash) + if status != warehouse.StatusOK { + info.status = DownloadCanceled + return + } + + info.file.Size = uint64(fileInfo.Size()) + info.status = DownloadActive + + // read the file + status, bytesRead, _ := core.UserWarehouse.ReadFile(info.hash, 0, int64(fileInfo.Size()), info.DiskFile.Handle) + + info.DiskFile.StoredSize = uint64(bytesRead) + + if status != warehouse.StatusOK { + info.status = DownloadCanceled + return + } + + info.Finish() + info.DeleteDefer(time.Hour * 1) // cache the details for 1 hour before removing} +} diff --git a/webapi/Search Dispatch.go b/webapi/Search Dispatch.go index 1b1414d..c62d69f 100644 --- a/webapi/Search Dispatch.go +++ b/webapi/Search Dispatch.go @@ -54,7 +54,10 @@ resultLoop: } } - if peer := core.NodelistLookup(file.NodeID); peer != nil { + if bytes.Equal(file.NodeID, core.SelfNodeID()) { + // Indicates data from the current user. + file.Tags = append(file.Tags, blockchain.TagFromNumber(blockchain.TagSharedByCount, 1)) + } else if peer := core.NodelistLookup(file.NodeID); peer != nil { // add the tags 'Shared By Count' and 'Shared By GeoIP' file.Tags = append(file.Tags, blockchain.TagFromNumber(blockchain.TagSharedByCount, 1)) if latitude, longitude, valid := api.Peer2GeoIP(peer); valid {