package db import ( "bytes" "encoding/hex" "fmt" "log" "math" "os" "sort" "strings" "time" "github.com/lbryio/hub/db/prefixes" "github.com/lbryio/lbry.go/v2/extras/util" lbryurl "github.com/lbryio/lbry.go/v2/url" "github.com/linxGnu/grocksdb" ) // // Constants // const ( nOriginalClaimExpirationTime = 262974 nExtendedClaimExpirationTime = 2102400 nExtendedClaimExpirationForkHeight = 400155 nNormalizedNameForkHeight = 539940 // targeting 21 March 2019 nMinTakeoverWorkaroundHeight = 496850 nMaxTakeoverWorkaroundHeight = 658300 // targeting 30 Oct 2019 nWitnessForkHeight = 680770 // targeting 11 Dec 2019 nAllClaimsInMerkleForkHeight = 658310 // targeting 30 Oct 2019 proportionalDelayFactor = 32 maxTakeoverDelay = 4032 ) // // Types and constructors, getters, setters, etc. // type ReadOnlyDBColumnFamily struct { DB *grocksdb.DB Handles map[string]*grocksdb.ColumnFamilyHandle Opts *grocksdb.ReadOptions TxCounts []uint32 Height uint32 BlockedStreams map[string][]byte BlockedChannels map[string][]byte FilteredStreams map[string][]byte FilteredChannels map[string][]byte } type ResolveResult struct { Name string NormalizedName string ClaimHash []byte TxNum uint32 Position uint16 TxHash []byte Height uint32 Amount uint64 ShortUrl string IsControlling bool CanonicalUrl string CreationHeight uint32 ActivationHeight uint32 ExpirationHeight uint32 EffectiveAmount uint64 SupportAmount uint64 Reposted int LastTakeoverHeight uint32 ClaimsInChannel uint32 ChannelHash []byte RepostedClaimHash []byte SignatureValid bool } type ResolveError struct { Error error } type OptionalResolveResultOrError interface { GetResult() *ResolveResult GetError() *ResolveError } type optionalResolveResultOrError struct { res *ResolveResult err *ResolveError } func (x *optionalResolveResultOrError) GetResult() *ResolveResult { return x.res } func (x *optionalResolveResultOrError) GetError() *ResolveError { return x.err } type ExpandedResolveResult struct { Stream OptionalResolveResultOrError Channel OptionalResolveResultOrError Repost OptionalResolveResultOrError RepostedChannel OptionalResolveResultOrError } type IterOptions struct { FillCache bool Prefix []byte Start []byte //interface{} Stop []byte //interface{} IncludeStart bool IncludeStop bool IncludeKey bool IncludeValue bool RawKey bool RawValue bool CfHandle *grocksdb.ColumnFamilyHandle It *grocksdb.Iterator } // NewIterateOptions creates a defualt options structure for a db iterator. func NewIterateOptions() *IterOptions { return &IterOptions{ FillCache: false, Prefix: []byte{}, Start: nil, Stop: nil, IncludeStart: true, IncludeStop: false, IncludeKey: true, IncludeValue: false, RawKey: false, RawValue: false, CfHandle: nil, It: nil, } } func (o *IterOptions) WithCfHandle(cfHandle *grocksdb.ColumnFamilyHandle) *IterOptions { o.CfHandle = cfHandle return o } func (o *IterOptions) WithFillCache(fillCache bool) *IterOptions { o.FillCache = fillCache return o } func (o *IterOptions) WithPrefix(prefix []byte) *IterOptions { o.Prefix = prefix return o } func (o *IterOptions) WithStart(start []byte) *IterOptions { o.Start = start return o } func (o *IterOptions) WithStop(stop []byte) *IterOptions { o.Stop = stop return o } func (o *IterOptions) WithIncludeStart(includeStart bool) *IterOptions { o.IncludeStart = includeStart return o } func (o *IterOptions) WithIncludeStop(includeStop bool) *IterOptions { o.IncludeStop = includeStop return o } func (o *IterOptions) WithIncludeKey(includeKey bool) *IterOptions { o.IncludeKey = includeKey return o } func (o *IterOptions) WithIncludeValue(includeValue bool) *IterOptions { o.IncludeValue = includeValue return o } func (o *IterOptions) WithRawKey(rawKey bool) *IterOptions { o.RawKey = rawKey return o } func (o *IterOptions) WithRawValue(rawValue bool) *IterOptions { o.RawValue = rawValue return o } type PathSegment struct { name string claimId string amountOrder int } func (ps *PathSegment) Normalized() string { return util.NormalizeName(ps.name) } func (ps *PathSegment) IsShortId() bool { return ps.claimId != "" && len(ps.claimId) < 40 } func (ps *PathSegment) IsFullId() bool { return len(ps.claimId) == 40 } func (ps *PathSegment) String() string { if ps.claimId != "" { return fmt.Sprintf("%s:%s", ps.name, ps.claimId) } else if ps.amountOrder != 0 { return fmt.Sprintf("%s:%d", ps.name, ps.amountOrder) } return ps.name } // BisectRight returns the index of the first element in the list that is greater than or equal to the value. func BisectRight(arr []uint32, val uint32) uint32 { i := sort.Search(len(arr), func(i int) bool { return arr[i] >= val }) return uint32(i) } func GetExpirationHeight(lastUpdatedHeight uint32) uint32 { return GetExpirationHeightFull(lastUpdatedHeight, false) } func GetExpirationHeightFull(lastUpdatedHeight uint32, extended bool) uint32 { if extended { return lastUpdatedHeight + nExtendedClaimExpirationTime } if lastUpdatedHeight < nExtendedClaimExpirationForkHeight { return lastUpdatedHeight + nOriginalClaimExpirationTime } return lastUpdatedHeight + nExtendedClaimExpirationTime } // // DB Get functions // func GetClaimsInChannelCount(db *ReadOnlyDBColumnFamily, channelHash []byte) (uint32, error) { prefix := []byte{prefixes.ChannelCount} handle := db.Handles[string(prefix)] key := prefixes.NewChannelCountKey(channelHash) rawKey := key.PackKey() slice, err := db.DB.GetCF(db.Opts, handle, rawKey) if err != nil { return 0, err } else if slice.Size() == 0 { return 0, nil } rawValue := make([]byte, len(slice.Data())) copy(rawValue, slice.Data()) value := prefixes.ChannelCountValueUnpack(rawValue) return value.Count, nil } func GetShortClaimIdUrl(db *ReadOnlyDBColumnFamily, name string, normalizedName string, claimHash []byte, rootTxNum uint32, rootPosition uint16) (string, error) { prefix := []byte{prefixes.ClaimShortIdPrefix} handle := db.Handles[string(prefix)] claimId := hex.EncodeToString(claimHash) claimIdLen := len(claimId) for prefixLen := 0; prefixLen < 10; prefixLen++ { var j int = prefixLen + 1 if j > claimIdLen { j = claimIdLen } partialClaimId := claimId[:j] partialKey := prefixes.NewClaimShortIDKey(normalizedName, partialClaimId) keyPrefix := prefixes.ClaimShortIDKeyPackPartial(partialKey, 2) // Prefix and handle options := NewIterateOptions().WithPrefix(prefix).WithCfHandle(handle) // Start and stop bounds options = options.WithStart(keyPrefix) // Don't include the key options = options.WithIncludeValue(false) ch := IterCF(db.DB, options) row := <-ch key := row.Key.(*prefixes.ClaimShortIDKey) if key.RootTxNum == rootTxNum && key.RootPosition == rootPosition { return fmt.Sprintf("%s#%s", name, key.PartialClaimId), nil } } return "", nil } func GetRepost(db *ReadOnlyDBColumnFamily, claimHash []byte) ([]byte, error) { prefix := []byte{prefixes.Repost} handle := db.Handles[string(prefix)] key := prefixes.NewRepostKey(claimHash) rawKey := key.PackKey() slice, err := db.DB.GetCF(db.Opts, handle, rawKey) if err != nil { return nil, err } else if slice.Size() == 0 { return nil, nil } rawValue := make([]byte, len(slice.Data())) copy(rawValue, slice.Data()) value := prefixes.RepostValueUnpack(rawValue) return value.RepostedClaimHash, nil } func GetRepostedCount(db *ReadOnlyDBColumnFamily, claimHash []byte) (int, error) { prefix := []byte{prefixes.RepostedClaim} handle := db.Handles[string(prefix)] key := prefixes.NewRepostedKey(claimHash) keyPrefix := prefixes.RepostedKeyPackPartial(key, 1) // Prefix and handle options := NewIterateOptions().WithPrefix(prefix).WithCfHandle(handle) // Start and stop bounds options = options.WithStart(keyPrefix) // Don't include the key options = options.WithIncludeValue(false) var i int = 0 ch := IterCF(db.DB, options) for range ch { i++ } return i, nil } func GetChannelForClaim(db *ReadOnlyDBColumnFamily, claimHash []byte, txNum uint32, position uint16) ([]byte, error) { key := prefixes.NewClaimToChannelKey(claimHash, txNum, position) cfName := string(prefixes.ClaimToChannel) handle := db.Handles[cfName] rawKey := key.PackKey() slice, err := db.DB.GetCF(db.Opts, handle, rawKey) if err != nil { return nil, err } else if slice.Size() == 0 { return nil, nil } rawValue := make([]byte, len(slice.Data())) copy(rawValue, slice.Data()) value := prefixes.ClaimToChannelValueUnpack(rawValue) return value.SigningHash, nil } func GetActiveAmount(db *ReadOnlyDBColumnFamily, claimHash []byte, txoType uint8, height uint32) (uint64, error) { cfName := string(prefixes.ActiveAmount) handle := db.Handles[cfName] startKey := &prefixes.ActiveAmountKey{ Prefix: []byte{prefixes.ActiveAmount}, ClaimHash: claimHash, TxoType: txoType, ActivationHeight: 0, } endKey := &prefixes.ActiveAmountKey{ Prefix: []byte{prefixes.ActiveAmount}, ClaimHash: claimHash, TxoType: txoType, ActivationHeight: height, } startKeyRaw := prefixes.ActiveAmountKeyPackPartial(startKey, 3) endKeyRaw := prefixes.ActiveAmountKeyPackPartial(endKey, 3) // Prefix and handle options := NewIterateOptions().WithPrefix([]byte{prefixes.ActiveAmount}).WithCfHandle(handle) // Start and stop bounds options = options.WithStart(startKeyRaw).WithStop(endKeyRaw) // Don't include the key options = options.WithIncludeKey(false).WithIncludeValue(true) ch := IterCF(db.DB, options) var sum uint64 = 0 for kv := range ch { sum += kv.Value.(*prefixes.ActiveAmountValue).Amount } return sum, nil } func GetEffectiveAmount(db *ReadOnlyDBColumnFamily, claimHash []byte, supportOnly bool) (uint64, error) { supportAmount, err := GetActiveAmount(db, claimHash, prefixes.ACTIVATED_SUPPORT_TXO_TYPE, db.Height+1) if err != nil { return 0, err } if supportOnly { return supportAmount, nil } activationAmount, err := GetActiveAmount(db, claimHash, prefixes.ACTIVATED_CLAIM_TXO_TYPE, db.Height+1) if err != nil { return 0, err } return activationAmount + supportAmount, nil } func GetSupportAmount(db *ReadOnlyDBColumnFamily, claimHash []byte) (uint64, error) { key := prefixes.NewSupportAmountKey(claimHash) cfName := string(prefixes.SupportAmount) handle := db.Handles[cfName] rawKey := key.PackKey() slice, err := db.DB.GetCF(db.Opts, handle, rawKey) if err != nil { return 0, err } else if slice.Size() == 0 { return 0, nil } rawValue := make([]byte, len(slice.Data())) copy(rawValue, slice.Data()) value := prefixes.SupportAmountValueUnpack(rawValue) return value.Amount, nil } func GetTxHash(db *ReadOnlyDBColumnFamily, txNum uint32) ([]byte, error) { /* if self._cache_all_tx_hashes: return self.total_transactions[tx_num] return self.prefix_db.tx_hash.get(tx_num, deserialize_value=False) */ // TODO: caching key := prefixes.NewTxHashKey(txNum) cfName := string(prefixes.TxHash) handle := db.Handles[cfName] rawKey := key.PackKey() slice, err := db.DB.GetCF(db.Opts, handle, rawKey) if err != nil { return nil, err } if slice.Size() == 0 { return nil, nil } rawValue := make([]byte, len(slice.Data())) copy(rawValue, slice.Data()) return rawValue, nil } func GetActivation(db *ReadOnlyDBColumnFamily, txNum uint32, postition uint16) (uint32, error) { return GetActivationFull(db, txNum, postition, false) } func GetActivationFull(db *ReadOnlyDBColumnFamily, txNum uint32, postition uint16, isSupport bool) (uint32, error) { var typ uint8 if isSupport { typ = prefixes.ACTIVATED_SUPPORT_TXO_TYPE } else { typ = prefixes.ACTIVATED_CLAIM_TXO_TYPE } key := prefixes.NewActivationKey(typ, txNum, postition) cfName := string(prefixes.ActivatedClaimAndSupport) handle := db.Handles[cfName] rawKey := key.PackKey() slice, err := db.DB.GetCF(db.Opts, handle, rawKey) if err != nil { return 0, err } rawValue := make([]byte, len(slice.Data())) copy(rawValue, slice.Data()) value := prefixes.ActivationValueUnpack(rawValue) // Does this need to explicitly return an int64, in case the uint32 overflows the max of an int? return value.Height, nil } func GetCachedClaimTxo(db *ReadOnlyDBColumnFamily, claim []byte) (*prefixes.ClaimToTXOValue, error) { // TODO: implement cache key := prefixes.NewClaimToTXOKey(claim) cfName := string(prefixes.ClaimToTXO) handle := db.Handles[cfName] rawKey := key.PackKey() slice, err := db.DB.GetCF(db.Opts, handle, rawKey) if err != nil { return nil, err } if slice.Size() == 0 { return nil, nil } rawValue := make([]byte, len(slice.Data())) copy(rawValue, slice.Data()) value := prefixes.ClaimToTXOValueUnpack(rawValue) return value, nil } func GetControllingClaim(db *ReadOnlyDBColumnFamily, name string) (*prefixes.ClaimTakeoverValue, error) { key := prefixes.NewClaimTakeoverKey(name) cfName := string(prefixes.ClaimTakeover) handle := db.Handles[cfName] rawKey := key.PackKey() slice, err := db.DB.GetCF(db.Opts, handle, rawKey) if err != nil { return nil, err } if slice.Size() == 0 { return nil, nil } rawValue := make([]byte, len(slice.Data())) copy(rawValue, slice.Data()) value := prefixes.ClaimTakeoverValueUnpack(rawValue) return value, nil } func FsGetClaimByHash(db *ReadOnlyDBColumnFamily, claimHash []byte) (*ResolveResult, error) { claim, err := GetCachedClaimTxo(db, claimHash) if err != nil { return nil, err } activation, err := GetActivation(db, claim.TxNum, claim.Position) if err != nil { return nil, err } log.Printf("%#v\n%#v\n%#v\n", claim, hex.EncodeToString(claimHash), activation) return PrepareResolveResult( db, claim.TxNum, claim.Position, claimHash, claim.Name, claim.RootTxNum, claim.RootPosition, activation, claim.ChannelSignatureIsValid, ) } func GetDBState(db *ReadOnlyDBColumnFamily) (*prefixes.DBStateValue, error) { prefix := []byte{prefixes.DBState} key := prefixes.NewDBStateKey() handle := db.Handles[string(prefix)] rawKey := key.PackKey() slice, err := db.DB.GetCF(db.Opts, handle, rawKey) if err != nil { return nil, err } else if slice.Size() == 0 { return nil, nil } rawValue := make([]byte, len(slice.Data())) copy(rawValue, slice.Data()) value := prefixes.DBStateValueUnpack(rawValue) return value, nil } func ClaimShortIdIter(db *ReadOnlyDBColumnFamily, normalizedName string, claimId string) <-chan *prefixes.PrefixRowKV { prefix := []byte{prefixes.ClaimShortIdPrefix} handle := db.Handles[string(prefix)] key := prefixes.NewClaimShortIDKey(normalizedName, claimId) var rawKeyPrefix []byte = nil if claimId != "" { rawKeyPrefix = prefixes.ClaimShortIDKeyPackPartial(key, 2) } else { rawKeyPrefix = prefixes.ClaimShortIDKeyPackPartial(key, 1) } options := NewIterateOptions().WithCfHandle(handle).WithPrefix(rawKeyPrefix) options = options.WithIncludeValue(true) //.WithIncludeStop(true) ch := IterCF(db.DB, options) return ch } func GetCachedClaimHash(db *ReadOnlyDBColumnFamily, txNum uint32, position uint16) (*prefixes.TXOToClaimValue, error) { // TODO: implement cache key := prefixes.NewTXOToClaimKey(txNum, position) cfName := string(prefixes.TXOToClaim) handle := db.Handles[cfName] rawKey := key.PackKey() slice, err := db.DB.GetCF(db.Opts, handle, rawKey) if err != nil { return nil, err } else if slice.Size() == 0 { return nil, nil } rawValue := make([]byte, len(slice.Data())) copy(rawValue, slice.Data()) value := prefixes.TXOToClaimValueUnpack(rawValue) return value, nil } // GetBlockerHash get the hash of the blocker or filterer of the claim. // TODO: this currently converts the byte arrays to strings, which is not // very efficient. Might want to figure out a better way to do this. func GetBlockerHash(db *ReadOnlyDBColumnFamily, claimHash, repostedClaimHash, channelHash []byte) ([]byte, []byte, error) { claimHashStr := string(claimHash) respostedClaimHashStr := string(repostedClaimHash) channelHashStr := string(channelHash) var blockedHash []byte = nil var filteredHash []byte = nil blockedHash = db.BlockedStreams[claimHashStr] if blockedHash == nil { blockedHash = db.BlockedStreams[respostedClaimHashStr] } if blockedHash == nil { blockedHash = db.BlockedChannels[claimHashStr] } if blockedHash == nil { blockedHash = db.BlockedChannels[respostedClaimHashStr] } if blockedHash == nil { blockedHash = db.BlockedChannels[channelHashStr] } filteredHash = db.FilteredStreams[claimHashStr] if filteredHash == nil { filteredHash = db.FilteredStreams[respostedClaimHashStr] } if filteredHash == nil { filteredHash = db.FilteredChannels[claimHashStr] } if filteredHash == nil { filteredHash = db.FilteredChannels[respostedClaimHashStr] } if filteredHash == nil { filteredHash = db.FilteredChannels[channelHashStr] } return blockedHash, filteredHash, nil } // // Resolve functions // // PrepareResolveResult prepares a ResolveResult to return func PrepareResolveResult( db *ReadOnlyDBColumnFamily, txNum uint32, position uint16, claimHash []byte, name string, rootTxNum uint32, rootPosition uint16, activationHeight uint32, signatureValid bool) (*ResolveResult, error) { normalizedName := util.NormalizeName(name) controllingClaim, err := GetControllingClaim(db, normalizedName) if err != nil { return nil, err } txHash, err := GetTxHash(db, txNum) if err != nil { return nil, err } height := BisectRight(db.TxCounts, txNum) createdHeight := BisectRight(db.TxCounts, rootTxNum) lastTakeoverHeight := controllingClaim.Height expirationHeight := GetExpirationHeight(height) supportAmount, err := GetSupportAmount(db, claimHash) if err != nil { return nil, err } claimToTxo, err := GetCachedClaimTxo(db, claimHash) if err != nil { return nil, err } claimAmount := claimToTxo.Amount effectiveAmount, err := GetEffectiveAmount(db, claimHash, false) if err != nil { return nil, err } channelHash, err := GetChannelForClaim(db, claimHash, txNum, position) if err != nil { return nil, err } repostedClaimHash, err := GetRepost(db, claimHash) if err != nil { return nil, err } shortUrl, err := GetShortClaimIdUrl(db, name, normalizedName, claimHash, txNum, rootPosition) if err != nil { return nil, err } var canonicalUrl string = shortUrl claimsInChannel, err := GetClaimsInChannelCount(db, claimHash) if err != nil { return nil, err } if channelHash != nil { // Ignore error because we already have this set if this doesn't work channelVals, _ := GetCachedClaimTxo(db, channelHash) log.Printf("channelVals: %#v\n", channelVals) if channelVals != nil { channelShortUrl, _ := GetShortClaimIdUrl( db, channelVals.Name, channelVals.NormalizedName(), channelHash, channelVals.RootTxNum, channelVals.RootPosition, ) canonicalUrl = fmt.Sprintf("%s/%s", channelShortUrl, shortUrl) } } reposted, err := GetRepostedCount(db, claimHash) if err != nil { return nil, err } isControlling := bytes.Equal(controllingClaim.ClaimHash, claimHash) return &ResolveResult{ Name: name, NormalizedName: normalizedName, ClaimHash: claimHash, TxNum: txNum, Position: position, TxHash: txHash, Height: height, Amount: claimAmount, ShortUrl: shortUrl, IsControlling: isControlling, CanonicalUrl: canonicalUrl, CreationHeight: createdHeight, ActivationHeight: activationHeight, ExpirationHeight: expirationHeight, EffectiveAmount: effectiveAmount, SupportAmount: supportAmount, Reposted: reposted, LastTakeoverHeight: lastTakeoverHeight, ClaimsInChannel: claimsInChannel, ChannelHash: channelHash, RepostedClaimHash: repostedClaimHash, SignatureValid: signatureValid, }, nil } func ResolveParsedUrl(db *ReadOnlyDBColumnFamily, parsed *PathSegment) (*ResolveResult, error) { normalizedName := util.NormalizeName(parsed.name) if (parsed.amountOrder == -1 && parsed.claimId == "") || parsed.amountOrder == 1 { controlling, err := GetControllingClaim(db, normalizedName) if err != nil { return nil, err } if controlling == nil { return nil, nil } return FsGetClaimByHash(db, controlling.ClaimHash) } var amountOrder int = int(math.Max(float64(parsed.amountOrder), 1)) log.Println("amountOrder:", amountOrder) if parsed.claimId != "" { if len(parsed.claimId) == 40 { claimHash, err := hex.DecodeString(parsed.claimId) if err != nil { return nil, err } // Maybe don't use caching version, when I actually implement the cache claimTxo, err := GetCachedClaimTxo(db, claimHash) if err != nil { return nil, err } if claimTxo == nil || claimTxo.NormalizedName() != normalizedName { return nil, nil } activation, err := GetActivation(db, claimTxo.TxNum, claimTxo.Position) if err != nil { return nil, err } return PrepareResolveResult( db, claimTxo.TxNum, claimTxo.Position, claimHash, claimTxo.Name, claimTxo.RootTxNum, claimTxo.RootPosition, activation, claimTxo.ChannelSignatureIsValid, ) } log.Println("nomalizedName:", normalizedName) log.Println("claimId:", parsed.claimId) var j int = 10 if len(parsed.claimId) < j { j = len(parsed.claimId) } ch := ClaimShortIdIter(db, normalizedName, parsed.claimId[:j]) row := <-ch key := row.Key.(*prefixes.ClaimShortIDKey) claimTxo := row.Value.(*prefixes.ClaimShortIDValue) fullClaimHash, err := GetCachedClaimHash(db, claimTxo.TxNum, claimTxo.Position) if err != nil { return nil, err } c, err := GetCachedClaimTxo(db, fullClaimHash.ClaimHash) if err != nil { return nil, err } nonNormalizedName := c.Name signatureIsValid := c.ChannelSignatureIsValid activation, err := GetActivation(db, claimTxo.TxNum, claimTxo.Position) if err != nil { return nil, err } return PrepareResolveResult( db, claimTxo.TxNum, claimTxo.Position, fullClaimHash.ClaimHash, nonNormalizedName, key.RootTxNum, key.RootPosition, activation, signatureIsValid, ) } return nil, nil } func ResolveClaimInChannel(db *ReadOnlyDBColumnFamily, channelHash []byte, normalizedName string) (*ResolveResult, error) { prefix := []byte{prefixes.ChannelToClaim} handle := db.Handles[string(prefix)] key := prefixes.NewChannelToClaimKey(channelHash, normalizedName) rawKeyPrefix := prefixes.ChannelToClaimKeyPackPartial(key, 2) options := NewIterateOptions().WithCfHandle(handle).WithPrefix(rawKeyPrefix) options = options.WithIncludeValue(true) //.WithIncludeStop(true) ch := IterCF(db.DB, options) // TODO: what's a good default size for this? var candidates []*ResolveResult = make([]*ResolveResult, 0, 100) var i = 0 for row := range ch { key := row.Key.(*prefixes.ChannelToClaimKey) stream := row.Value.(*prefixes.ChannelToClaimValue) effectiveAmount, err := GetEffectiveAmount(db, stream.ClaimHash, false) if err != nil { return nil, err } if i == 0 || candidates[i-1].Amount == effectiveAmount { candidates = append( candidates, &ResolveResult{ TxNum: key.TxNum, Position: key.Position, ClaimHash: stream.ClaimHash, Amount: effectiveAmount, ChannelHash: channelHash, NormalizedName: normalizedName, }, ) i++ } else { break } } log.Printf("candidates: %#v\n", candidates) if len(candidates) == 0 { return nil, nil } else { // return list(sorted(candidates, key=lambda item: item[1]))[0] sort.Slice(candidates, func(i, j int) bool { return candidates[i].Amount < candidates[j].Amount }) return candidates[0], nil } } func Resolve(db *ReadOnlyDBColumnFamily, url string) *ExpandedResolveResult { var res = &ExpandedResolveResult{ Stream: nil, Channel: nil, Repost: nil, RepostedChannel: nil, } var channel *PathSegment = nil var stream *PathSegment = nil parsed, err := lbryurl.Parse(url, false) if err != nil { res.Stream = &optionalResolveResultOrError{ err: &ResolveError{err}, } return res } log.Printf("parsed: %#v\n", parsed) // has stream in channel if strings.Compare(parsed.StreamName, "") != 0 && strings.Compare(parsed.ClaimName, "") != 0 { channel = &PathSegment{ name: parsed.ClaimName, claimId: parsed.ChannelClaimId, amountOrder: parsed.PrimaryBidPosition, } stream = &PathSegment{ name: parsed.StreamName, claimId: parsed.StreamClaimId, amountOrder: parsed.SecondaryBidPosition, } } else if strings.Compare(parsed.ClaimName, "") != 0 { channel = &PathSegment{ name: parsed.ClaimName, claimId: parsed.ChannelClaimId, amountOrder: parsed.PrimaryBidPosition, } } else if strings.Compare(parsed.StreamName, "") != 0 { stream = &PathSegment{ name: parsed.StreamName, claimId: parsed.StreamClaimId, amountOrder: parsed.SecondaryBidPosition, } } log.Printf("channel: %#v\n", channel) log.Printf("stream: %#v\n", stream) var resolvedChannel *ResolveResult = nil var resolvedStream *ResolveResult = nil if channel != nil { resolvedChannel, err = ResolveParsedUrl(db, channel) if err != nil { res.Channel = &optionalResolveResultOrError{ err: &ResolveError{err}, } return res } else if resolvedChannel == nil { res.Channel = &optionalResolveResultOrError{ err: &ResolveError{fmt.Errorf("could not find channel in \"%s\"", url)}, } return res } } log.Printf("resolvedChannel: %#v\n", resolvedChannel) log.Printf("resolvedChannel.TxHash: %s\n", hex.EncodeToString(resolvedChannel.TxHash)) log.Printf("resolvedChannel.ClaimHash: %s\n", hex.EncodeToString(resolvedChannel.ClaimHash)) log.Printf("resolvedChannel.ChannelHash: %s\n", hex.EncodeToString(resolvedChannel.ChannelHash)) log.Printf("stream %#v\n", stream) if stream != nil { if resolvedChannel != nil { streamClaim, err := ResolveClaimInChannel(db, resolvedChannel.ClaimHash, stream.Normalized()) log.Printf("streamClaim %#v\n", streamClaim) log.Printf("streamClaim.ClaimHash: %s\n", hex.EncodeToString(streamClaim.ClaimHash)) log.Printf("streamClaim.ChannelHash: %s\n", hex.EncodeToString(streamClaim.ChannelHash)) // TODO: Confirm error case if err != nil { res.Stream = &optionalResolveResultOrError{ err: &ResolveError{err}, } return res } if streamClaim != nil { resolvedStream, err = FsGetClaimByHash(db, streamClaim.ClaimHash) // TODO: Confirm error case if err != nil { res.Stream = &optionalResolveResultOrError{ err: &ResolveError{err}, } return res } } } else { resolvedStream, err = ResolveParsedUrl(db, stream) // TODO: Confirm error case if err != nil { res.Stream = &optionalResolveResultOrError{ err: &ResolveError{err}, } return res } if channel == nil && resolvedChannel == nil && resolvedStream != nil && len(resolvedStream.ChannelHash) > 0 { resolvedChannel, err = FsGetClaimByHash(db, resolvedStream.ChannelHash) // TODO: Confirm error case if err != nil { res.Channel = &optionalResolveResultOrError{ err: &ResolveError{err}, } return res } } } if resolvedStream == nil { res.Stream = &optionalResolveResultOrError{ err: &ResolveError{fmt.Errorf("could not find stream in \"%s\"", url)}, } return res } } // Getting blockers and filters var repost *ResolveResult = nil var repostedChannel *ResolveResult = nil log.Printf("about to get blockers and filters: %#v, %#v\n", resolvedChannel, resolvedStream) if resolvedStream != nil || resolvedChannel != nil { var claim *ResolveResult = nil var claimHash []byte = nil var respostedClaimHash []byte = nil var blockerHash []byte = nil if resolvedStream != nil { claim = resolvedStream claimHash = resolvedStream.ClaimHash respostedClaimHash = resolvedStream.RepostedClaimHash } else { claim = resolvedChannel claimHash = resolvedChannel.ClaimHash } blockerHash, _, err = GetBlockerHash(db, claimHash, respostedClaimHash, claim.ChannelHash) log.Printf("blockerHash: %s\n", hex.EncodeToString(blockerHash)) if err != nil { res.Channel = &optionalResolveResultOrError{ err: &ResolveError{err}, } return res } if blockerHash != nil { reasonRow, err := FsGetClaimByHash(db, blockerHash) if err != nil { res.Channel = &optionalResolveResultOrError{ err: &ResolveError{err}, } return res } res.Channel = &optionalResolveResultOrError{ err: &ResolveError{fmt.Errorf("%s, %v, %v", url, blockerHash, reasonRow)}, } return res } if claim.RepostedClaimHash != nil { repost, err = FsGetClaimByHash(db, claim.RepostedClaimHash) if err != nil { res.Channel = &optionalResolveResultOrError{ err: &ResolveError{err}, } return res } if repost != nil && repost.ChannelHash != nil && repost.SignatureValid { repostedChannel, err = FsGetClaimByHash(db, repost.ChannelHash) if err != nil { res.Channel = &optionalResolveResultOrError{ err: &ResolveError{err}, } return res } } } } res.Channel = &optionalResolveResultOrError{ res: resolvedChannel, } res.Stream = &optionalResolveResultOrError{ res: resolvedStream, } res.Repost = &optionalResolveResultOrError{ res: repost, } res.RepostedChannel = &optionalResolveResultOrError{ res: repostedChannel, } log.Printf("parsed: %+v\n", parsed) return res } // // Iterators / db construction functions // func (o *IterOptions) StopIteration(key []byte) bool { if key == nil { return false } maxLen := int(math.Min(float64(len(key)), float64(len(o.Stop)))) if o.Stop != nil && (bytes.HasPrefix(key, o.Stop) || bytes.Compare(o.Stop, key[:maxLen]) < 0) { return true } else if o.Start != nil && bytes.Compare(o.Start, key[:len(o.Start)]) > 0 { return true } else if o.Prefix != nil && !bytes.HasPrefix(key, o.Prefix) { return true } return false } func (opts *IterOptions) ReadRow(ch chan *prefixes.PrefixRowKV, prevKey *[]byte) bool { it := opts.It key := it.Key() keyData := key.Data() keyLen := len(keyData) value := it.Value() valueData := value.Data() valueLen := len(valueData) var outKey interface{} = nil var outValue interface{} = nil var err error = nil // log.Println("keyData:", keyData) // log.Println("valueData:", valueData) // We need to check the current key if we're not including the stop // key. if !opts.IncludeStop && opts.StopIteration(keyData) { log.Println("returning false") return false } // We have to copy the key no matter what because we need to check // it on the next iterations to see if we're going to stop. newKeyData := make([]byte, keyLen) copy(newKeyData, keyData) if opts.IncludeKey && !opts.RawKey { outKey, err = prefixes.UnpackGenericKey(newKeyData) if err != nil { log.Println(err) } } else if opts.IncludeKey { outKey = newKeyData } // Value could be quite large, so this setting could be important // for performance in some cases. if opts.IncludeValue { newValueData := make([]byte, valueLen) copy(newValueData, valueData) if !opts.RawValue { outValue, err = prefixes.UnpackGenericValue(newKeyData, newValueData) if err != nil { log.Println(err) } } else { outValue = newValueData } } key.Free() value.Free() // log.Println("sending to channel") ch <- &prefixes.PrefixRowKV{ Key: outKey, Value: outValue, } *prevKey = newKeyData // log.Println("*prevKey:", *prevKey) return true } func IterCF(db *grocksdb.DB, opts *IterOptions) <-chan *prefixes.PrefixRowKV { ch := make(chan *prefixes.PrefixRowKV) ro := grocksdb.NewDefaultReadOptions() ro.SetFillCache(opts.FillCache) it := db.NewIteratorCF(ro, opts.CfHandle) // it := db.NewIterator(ro) opts.It = it it.Seek(opts.Prefix) if opts.Start != nil { log.Println("Seeking to start") it.Seek(opts.Start) } go func() { defer it.Close() defer close(ch) var prevKey []byte = nil if !opts.IncludeStart { log.Println("Not including start") it.Next() } if !it.Valid() && opts.IncludeStop { log.Println("Not valid, but including stop") opts.ReadRow(ch, &prevKey) } var continueIter bool = true for ; continueIter && !opts.StopIteration(prevKey) && it.Valid(); it.Next() { //log.Println("Main loop") continueIter = opts.ReadRow(ch, &prevKey) } }() return ch } func Iter(db *grocksdb.DB, opts *IterOptions) <-chan *prefixes.PrefixRowKV { ch := make(chan *prefixes.PrefixRowKV) ro := grocksdb.NewDefaultReadOptions() /* FIXME: ro.SetIterateLowerBound() ro.SetIterateUpperBound() ro.PrefixSameAsStart() -> false ro.AutoPrefixMode() -> on */ ro.SetFillCache(opts.FillCache) it := db.NewIterator(ro) opts.It = it it.Seek(opts.Prefix) if opts.Start != nil { it.Seek(opts.Start) } go func() { defer it.Close() defer close(ch) var prevKey []byte = nil if !opts.IncludeStart { it.Next() } if !it.Valid() && opts.IncludeStop { opts.ReadRow(ch, &prevKey) } for ; !opts.StopIteration(prevKey) && it.Valid(); it.Next() { opts.ReadRow(ch, &prevKey) } }() return ch } func GetWriteDBCF(name string) (*grocksdb.DB, []*grocksdb.ColumnFamilyHandle, error) { opts := grocksdb.NewDefaultOptions() cfOpt := grocksdb.NewDefaultOptions() cfNames, err := grocksdb.ListColumnFamilies(opts, name) if err != nil { return nil, nil, err } cfOpts := make([]*grocksdb.Options, len(cfNames)) for i := range cfNames { cfOpts[i] = cfOpt } db, handles, err := grocksdb.OpenDbColumnFamilies(opts, name, cfNames, cfOpts) if err != nil { return nil, nil, err } for i, handle := range handles { log.Printf("%d: %s, %+v\n", i, cfNames[i], handle) } return db, handles, nil } func GetDBColumnFamlies(name string, cfNames []string) (*ReadOnlyDBColumnFamily, error) { opts := grocksdb.NewDefaultOptions() roOpts := grocksdb.NewDefaultReadOptions() cfOpt := grocksdb.NewDefaultOptions() //cfNames := []string{"default", cf} cfOpts := make([]*grocksdb.Options, len(cfNames)) for i := range cfNames { cfOpts[i] = cfOpt } db, handles, err := grocksdb.OpenDbAsSecondaryColumnFamilies(opts, name, "asdf", cfNames, cfOpts) // db, handles, err := grocksdb.OpenDbColumnFamilies(opts, name, cfNames, cfOpts) if err != nil { return nil, err } var handlesMap = make(map[string]*grocksdb.ColumnFamilyHandle) for i, handle := range handles { log.Printf("%d: %+v\n", i, handle) handlesMap[cfNames[i]] = handle } myDB := &ReadOnlyDBColumnFamily{ DB: db, Handles: handlesMap, Opts: roOpts, BlockedStreams: make(map[string][]byte), BlockedChannels: make(map[string][]byte), FilteredStreams: make(map[string][]byte), FilteredChannels: make(map[string][]byte), TxCounts: nil, Height: 0, } err = InitTxCounts(myDB) if err != nil { return nil, err } return myDB, nil } // DetectChanges keep the rocksdb db in sync func DetectChanges(db *ReadOnlyDBColumnFamily) error { err := db.DB.TryCatchUpWithPrimary() if err != nil { log.Printf("error trying to catch up with primary: %#v", err) return err } return nil } func InitTxCounts(db *ReadOnlyDBColumnFamily) error { start := time.Now() handle, ok := db.Handles[string([]byte{prefixes.TxCount})] if !ok { return fmt.Errorf("TxCount prefix not found") } //TODO: figure out a reasonable default and make it a constant txCounts := make([]uint32, 0, 1200000) options := NewIterateOptions().WithPrefix([]byte{prefixes.TxCount}).WithCfHandle(handle) options = options.WithIncludeKey(false).WithIncludeValue(true) ch := IterCF(db.DB, options) for txCount := range ch { txCounts = append(txCounts, txCount.Value.(*prefixes.TxCountValue).TxCount) } duration := time.Since(start) log.Println("len(txCounts), cap(txCounts):", len(txCounts), cap(txCounts)) log.Println("Time to get txCounts:", duration) db.TxCounts = txCounts db.Height = uint32(len(txCounts)) return nil } func GetDBCF(name string, cf string) (*grocksdb.DB, []*grocksdb.ColumnFamilyHandle, error) { opts := grocksdb.NewDefaultOptions() cfOpt := grocksdb.NewDefaultOptions() cfNames := []string{"default", cf} cfOpts := []*grocksdb.Options{cfOpt, cfOpt} db, handles, err := grocksdb.OpenDbAsSecondaryColumnFamilies(opts, name, "asdf", cfNames, cfOpts) for i, handle := range handles { log.Printf("%d: %+v\n", i, handle) } if err != nil { return nil, nil, err } return db, handles, nil } func GetDB(name string) (*grocksdb.DB, error) { opts := grocksdb.NewDefaultOptions() db, err := grocksdb.OpenDbAsSecondary(opts, name, "asdf") if err != nil { return nil, err } return db, nil } func ReadPrefixN(db *grocksdb.DB, prefix []byte, n int) []*prefixes.PrefixRowKV { ro := grocksdb.NewDefaultReadOptions() ro.SetFillCache(false) it := db.NewIterator(ro) defer it.Close() res := make([]*prefixes.PrefixRowKV, n) var i = 0 it.Seek(prefix) for ; it.Valid(); it.Next() { key := it.Key() value := it.Value() res[i] = &prefixes.PrefixRowKV{ Key: key.Data(), Value: value.Data(), } key.Free() value.Free() i++ if i >= n { break } } return res } func ReadWriteRawNColumnFamilies(db *grocksdb.DB, options *IterOptions, out string, n int) { options.RawKey = true options.RawValue = true ch := IterCF(db, options) file, err := os.Create(out) if err != nil { log.Println(err) return } defer file.Close() var i = 0 log.Println(options.Prefix) cf := string(options.Prefix) file.Write([]byte(fmt.Sprintf("%s,,\n", options.Prefix))) for kv := range ch { log.Println(i) if i >= n { return } key := kv.Key.([]byte) value := kv.Value.([]byte) keyHex := hex.EncodeToString(key) valueHex := hex.EncodeToString(value) //log.Println(keyHex) //log.Println(valueHex) file.WriteString(cf) file.WriteString(",") file.WriteString(keyHex) file.WriteString(",") file.WriteString(valueHex) file.WriteString("\n") i++ } } func ReadWriteRawNCF(db *grocksdb.DB, options *IterOptions, out string, n int) { options.RawKey = true options.RawValue = true ch := IterCF(db, options) file, err := os.Create(out) if err != nil { log.Println(err) return } defer file.Close() var i = 0 log.Println(options.Prefix) file.Write([]byte(fmt.Sprintf("%s,\n", options.Prefix))) for kv := range ch { log.Println(i) if i >= n { return } key := kv.Key.([]byte) value := kv.Value.([]byte) keyHex := hex.EncodeToString(key) valueHex := hex.EncodeToString(value) //log.Println(keyHex) //log.Println(valueHex) file.WriteString(keyHex) file.WriteString(",") file.WriteString(valueHex) file.WriteString("\n") i++ } } func ReadWriteRawN(db *grocksdb.DB, options *IterOptions, out string, n int) { options.RawKey = true options.RawValue = true ch := Iter(db, options) file, err := os.Create(out) if err != nil { log.Println(err) return } defer file.Close() var i = 0 for kv := range ch { log.Println(i) if i >= n { return } key := kv.Key.([]byte) value := kv.Value.([]byte) keyHex := hex.EncodeToString(key) valueHex := hex.EncodeToString(value) log.Println(keyHex) log.Println(valueHex) file.WriteString(keyHex) file.WriteString(",") file.WriteString(valueHex) file.WriteString("\n") i++ } } func GenerateTestData(prefix byte, fileName string) { dbVal, err := GetDB("/mnt/d/data/wallet/lbry-rocksdb/") if err != nil { log.Fatalln(err) } options := NewIterateOptions() options.WithRawKey(true).WithRawValue(true).WithIncludeValue(true) options.WithPrefix([]byte{prefix}) ReadWriteRawN(dbVal, options, fileName, 10) }