minTimestamp = dateFns.subDays(new Date(), 7)
ips = {
const ips = new Set()
for (const record of multiaddrsIpsReport.multiaddrsIps) {
if (d3.isoParse(record.timestamp) > new Date(minTimestamp)) {
ips.add(record.ip)
}
}
return ips
}
newIps = {
const newIps = new Set()
for (const ip of [...ips]) {
if (!latestIpsGeoLite2Report.ipsGeoLite2[ip]) {
newIps.add(ip)
}
}
return newIps
}
async function* lookupIpsStream() {
const concurrency = 15
const callGeoLookupsStream = transform(concurrency, async ip => {
try {
const geoLookup = await (await fetch(`${geoApiBaseUrl}/geolite2/${ip}`)).json()
return {
ip,
continent: geoLookup.continent ? geoLookup.continent.code : null,
country: geoLookup.country ? geoLookup.country.isoCode : null,
subdiv1: geoLookup.subdivisions && geoLookup.subdivisions.length > 0 ? geoLookup.subdivisions[0].isoCode : null,
city: geoLookup.city && geoLookup.city.names ? geoLookup.city.names.en : null,
long: geoLookup.location ? geoLookup.location.longitude : null,
lat: geoLookup.location ? geoLookup.location.latitude : null,
geolite2: geoLookup
}
} catch (e) {
console.info('IP lookup error', ip, e.message)
return {}
}
})
const startTime = new Date()
let counter = 0
let hits = 0
let errors = 0
for await (const geoLookup of callGeoLookupsStream([...newIps])) {
const now = new Date()
if (now - startTime > maxElapsed) {
yield {
done: true,
timeout: true,
counter,
hits,
errors
}
return
}
if (geoLookup.ip) {
hits++
yield {
counter,
hits,
errors,
...geoLookup
}
} else {
errors++
yield {
counter,
hits,
errors,
}
}
counter++
}
yield {
done: true,
counter,
hits,
errors
}
}
// lookupIpsStream()
ipsGeoLite2 = {
if (start === 0) {
yield {
state: 'paused'
}
return
}
yield {
state: 'starting'
}
let records = []
let totalErrors = 0
let lastCounter = 0
const startTime = new Date()
for await (const {counter, hits, errors, ...record} of lookupIpsStream()) {
totalErrors = errors
lastCounter = counter
if (record.ip) {
records.push(record)
}
yield {
state: "streaming",
elapsed: ((new Date()) - startTime) / 1000,
scannedIps: counter,
totalIps: ipsCount,
recordsLength: records.length,
errors
}
}
const endTime = new Date()
yield {
state: "done",
elapsed: (endTime - startTime) / 1000,
scannedIps: lastCounter,
totalIps: ipsCount,
records: records.sort(),
startTime,
endTime,
errors: totalErrors
}
}
import {dateToEpoch, epochToDate} from '@jbenet/filecoin-chain-time-calculator'
Purpose-built for displays of data
Observable is your go-to platform for exploring data and creating expressive data visualizations. Use reactive JavaScript notebooks for prototyping and a collaborative canvas for visual data exploration and dashboard creation.
