1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
|
package main
import (
"log"
"net"
"strings"
"time"
"github.com/getwtxt/registry"
"github.com/syndtr/goleveldb/leveldb"
)
// Checks whether it's time to refresh
// the cache.
func checkCacheTime() bool {
return time.Since(confObj.lastCache) > confObj.cacheInterval
}
// Checks whether it's time to push
// the cache to the database
func checkDBtime() bool {
return time.Since(confObj.lastPush) > confObj.dbInterval
}
// Launched by init as a goroutine to constantly watch
// for the update interval to pass.
func cacheAndPush() {
for {
if checkCacheTime() {
refreshCache()
}
if checkDBtime() {
if err := pushDatabase(); err != nil {
log.Printf("Error pushing cache to database: %v\n", err)
}
}
}
}
// Refreshes the cache.
func refreshCache() {
// Iterate over the registry and
// update each individual user.
for k := range twtxtCache.Reg {
err := twtxtCache.UpdateUser(k)
if err != nil {
log.Printf("%v\n", err)
continue
}
}
// Re-scrape all the remote registries
// to see if they have any new users
// to add locally.
for _, v := range remoteRegistries.List {
err := twtxtCache.ScrapeRemoteRegistry(v)
if err != nil {
log.Printf("Error while refreshing local copy of remote registry user data: %v\n", err)
}
}
confObj.mu.Lock()
confObj.lastCache = time.Now()
confObj.mu.Unlock()
}
// Pushes the registry's cache data to a local
// database for safe keeping.
func pushDatabase() error {
// Acquire the database from the aether.
// goleveldb is concurrency-safe, so we
// can immediately push it back into the
// channel for other functions to use.
db := <-dbChan
dbChan <- db
// Create a batch write job so it can
// be done at one time rather than
// per entry.
twtxtCache.Mu.RLock()
var dbBasket = &leveldb.Batch{}
for k, v := range twtxtCache.Reg {
dbBasket.Put([]byte(k+"*Nick"), []byte(v.Nick))
dbBasket.Put([]byte(k+"*URL"), []byte(v.URL))
dbBasket.Put([]byte(k+"*IP"), []byte(v.IP.String()))
dbBasket.Put([]byte(k+"*Date"), []byte(v.Date))
for i, e := range v.Status {
rfc := i.Format(time.RFC3339)
dbBasket.Put([]byte(k+"*Status*"+rfc), []byte(e))
}
}
twtxtCache.Mu.RUnlock()
// Save our list of remote registries to scrape.
for k, v := range remoteRegistries.List {
dbBasket.Put([]byte("remote*"+string(k)), []byte(v))
}
// Execute the batch job.
if err := db.Write(dbBasket, nil); err != nil {
return err
}
// Update the last push time for
// our timer/watch function to
// reference.
confObj.mu.Lock()
confObj.lastPush = time.Now()
confObj.mu.Unlock()
return nil
}
// Pulls registry data from the DB on startup.
// Iterates over the database one entry at a time.
func pullDatabase() {
// Acquire the database from the aether.
// goleveldb is concurrency-safe, so we
// can immediately push it back into the
// channel for other functions to use.
db := <-dbChan
dbChan <- db
iter := db.NewIterator(nil, nil)
// Read the database entry-by-entry
for iter.Next() {
key := string(iter.Key())
val := string(iter.Value())
split := strings.Split(key, "*")
urls := string(split[0])
field := string(split[1])
if urls != "remote" {
// Start with an empty Data struct. If
// there's already one in the cache, pull
// it and use it instead.
data := registry.NewUserData()
twtxtCache.Mu.RLock()
if _, ok := twtxtCache.Reg[urls]; ok {
data = twtxtCache.Reg[urls]
}
twtxtCache.Mu.RUnlock()
switch field {
case "IP":
data.IP = net.ParseIP(val)
case "Nick":
data.Nick = val
case "URL":
data.URL = val
case "Date":
data.Date = val
case "Status":
// If we're looking at a Status entry in the DB,
// parse the time then add it to the TimeMap under
// data.Status
thetime, err := time.Parse(time.RFC3339, split[2])
if err != nil {
log.Printf("%v\n", err)
}
data.Status[thetime] = val
}
// Push the data struct (back) into
// the cache.
twtxtCache.Mu.Lock()
twtxtCache.Reg[urls] = data
twtxtCache.Mu.Unlock()
} else {
// If we've come across an entry for
// a remote twtxt registry to scrape,
// add it to our list.
remoteRegistries.Mu.Lock()
remoteRegistries.List = append(remoteRegistries.List, val)
remoteRegistries.Mu.Unlock()
}
}
iter.Release()
err := iter.Error()
if err != nil {
log.Printf("Error while pulling DB into registry cache: %v\n", err)
}
}
|