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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204

package main

import (
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"
)

const (
	// fetchLimit is as much of a page as is kept, the same as
	// logic.ReadSiteMeta reads.
	fetchLimit = 1 << 20
	// hostPause is the wait between two requests to the same host.
	hostPause    = 2 * time.Second
	fetchTimeout = 20 * time.Second
	maxRedirects = 10
)

type job struct{ handle, url string }

// fetch gets the page of every sampled site not fetched yet. Each host is
// asked one request at a time, hostPause apart, and up to hosts hosts at
// once. A failed request is stored too, with its error, and is not retried:
// DELETE its row from fetch to try it again.
func fetch(db *sql.DB, hosts int, log io.Writer) error {
	rows, err := db.Query(`SELECT l.handle, l.site_url FROM sample s
		JOIN listing l ON l.handle=s.handle
		WHERE s.handle NOT IN (SELECT handle FROM fetch)
		ORDER BY s.rank`)
	if err != nil {
		return err
	}
	var jobs []job
	for rows.Next() {
		var j job
		if err = rows.Scan(&j.handle, &j.url); err != nil {
			rows.Close()
			return err
		}
		jobs = append(jobs, j)
	}
	rows.Close()
	if err = rows.Err(); err != nil {
		return err
	}
	return fetchAll(jobs, hosts, "pages", pageAccept, log, func(j job, r fetchRow) error {
		return storeFetch(db, "fetch", j, r)
	})
}

// pageAccept is what a browser asks for, roughly.
const pageAccept = "text/html,application/xhtml+xml;q=0.9,*/*;q=0.8"

// fetchAll gets the URL of every job and hands it to store. Each host is
// asked one request at a time, hostPause apart, and up to hosts hosts at
// once; what is fetched, what is logged as.
func fetchAll(jobs []job, hosts int, what, accept string, log io.Writer, store func(job, fetchRow) error) error {
	byHost := map[string][]job{}
	var order []string
	for _, j := range jobs {
		h := hostOf(j.url)
		if _, ok := byHost[h]; !ok {
			order = append(order, h)
		}
		byHost[h] = append(byHost[h], j)
	}
	total := len(jobs)
	fmt.Fprintf(log, "%d %s to fetch from %d hosts\n", total, what, len(order))

	client := &http.Client{
		Timeout: fetchTimeout,
		CheckRedirect: func(req *http.Request, via []*http.Request) error {
			if len(via) >= maxRedirects {
				return fmt.Errorf("stopped after %d redirects", maxRedirects)
			}
			return nil
		},
	}

	var mu sync.Mutex // guards done, firstErr and the log
	done := 0
	var firstErr error
	hostCh := make(chan string)
	var wg sync.WaitGroup
	for range max(hosts, 1) {
		wg.Go(func() {
			for h := range hostCh {
				for i, j := range byHost[h] {
					if i > 0 {
						time.Sleep(hostPause)
					}
					row := get(client, j.url, accept)
					err := store(j, row)
					mu.Lock()
					done++
					status := row.errText
					if status == "" {
						status = fmt.Sprintf("%d %s", row.status, row.contentType)
					}
					fmt.Fprintf(log, "[%d/%d] %s: %s\n", done, total, j.url, status)
					if err != nil && firstErr == nil {
						firstErr = err
					}
					mu.Unlock()
				}
			}
		})
	}
	for _, h := range order {
		hostCh <- h
	}
	close(hostCh)
	wg.Wait()
	return firstErr
}

type fetchRow struct {
	fetchedAt   time.Time
	status      int
	finalUrl    string
	contentType string
	headers     string
	body        []byte
	truncated   bool
	errText     string
	ms          int64
}

func get(client *http.Client, u, accept string) fetchRow {
	start := time.Now()
	row := fetchRow{fetchedAt: start.UTC()}
	defer func() { row.ms = time.Since(start).Milliseconds() }()

	req, err := http.NewRequest("GET", u, nil)
	if err != nil {
		row.errText = err.Error()
		return row
	}
	req.Header.Set("User-Agent", userAgent)
	req.Header.Set("Accept", accept)
	resp, err := client.Do(req)
	if err != nil {
		row.errText = err.Error()
		return row
	}
	defer resp.Body.Close()
	row.status = resp.StatusCode
	row.finalUrl = resp.Request.URL.String()
	row.contentType = resp.Header.Get("Content-Type")
	if h, err := json.Marshal(resp.Header); err == nil {
		row.headers = string(h)
	}
	// One byte more than is kept tells whether there was more.
	body, err := io.ReadAll(io.LimitReader(resp.Body, fetchLimit+1))
	if len(body) > fetchLimit {
		body, row.truncated = body[:fetchLimit], true
	}
	row.body = body
	if err != nil && !errors.Is(err, io.EOF) {
		row.errText = "reading body: " + err.Error()
	}
	return row
}

// storeFetch stores a response in table, fetch or feed_fetch, which have the
// same columns.
func storeFetch(db *sql.DB, table string, j job, r fetchRow) error {
	null := func(s string) any {
		if s == "" {
			return nil
		}
		return s
	}
	var status any
	if r.status != 0 {
		status = r.status
	}
	var truncated any
	if r.body != nil {
		truncated = r.truncated
	}
	_, err := db.Exec(`INSERT INTO `+table+`
		(handle, url, fetched_at, status, final_url, content_type, headers, body, truncated, error, ms)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		j.handle, j.url, r.fetchedAt.Format(time.RFC3339), status, null(r.finalUrl), null(r.contentType),
		null(r.headers), r.body, truncated, null(r.errText), r.ms)
	return err
}

// hostOf is the host a URL is fetched from, in lower case, to group the
// requests by.
func hostOf(s string) string {
	if u, err := url.Parse(strings.TrimSpace(s)); err == nil && u.Host != "" {
		return strings.ToLower(u.Hostname())
	}
	return s
}