You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
407 lines
9.8 KiB
407 lines
9.8 KiB
package deadlinks
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"code.betamike.com/mediocregopher/mediocre-go-lib/miter"
|
|
|
|
_ "github.com/mattn/go-sqlite3"
|
|
migrate "github.com/rubenv/sql-migrate"
|
|
)
|
|
|
|
// Store keeps track of the current status of all discovered Resources.
|
|
// Resources with no incoming links will be periodically cleaned out.
|
|
//
|
|
// An implementation of Store must be thread-safe.
|
|
type Store interface {
|
|
// GetByStatus returns all Resources with the given Status.
|
|
GetByStatus(ResourceStatus) miter.Iterator[Resource]
|
|
|
|
// GetURLsByLastChecked returns the URLs of all Resources with LastChecked
|
|
// values older than the given timestamp, or which have never been checked.
|
|
GetURLsByLastChecked(olderThan time.Time) miter.Iterator[URL]
|
|
|
|
// SetPinned overwrites the set of pinned URLs with the given one.
|
|
SetPinned(context.Context, []URL) error
|
|
|
|
// Update updates the Resource identified by the given URL with the given
|
|
// arguments. The Resource must have been Touch'd previously, or this
|
|
// returns an error.
|
|
Update(
|
|
ctx context.Context,
|
|
now time.Time,
|
|
url URL,
|
|
status ResourceStatus,
|
|
errorString string,
|
|
outgoing []URL,
|
|
) error
|
|
}
|
|
|
|
var migrations = &migrate.MemoryMigrationSource{Migrations: []*migrate.Migration{
|
|
{
|
|
Id: "1",
|
|
Up: []string{
|
|
`CREATE TABLE urls (
|
|
id INTEGER NOT NULL PRIMARY KEY,
|
|
url TEXT NOT NULL,
|
|
UNIQUE(url)
|
|
)`,
|
|
|
|
`CREATE TABLE resources (
|
|
url_id INTEGER NOT NULL PRIMARY KEY,
|
|
status INTEGER NOT NULL DEFAULT 0,
|
|
pinned INTEGER NOT NULL,
|
|
last_checked INTEGER NOT NULL DEFAULT 0,
|
|
error_string TEXT NOT NULL DEFAULT '',
|
|
FOREIGN KEY(url_id) REFERENCES urls(id) ON DELETE CASCADE
|
|
)`,
|
|
|
|
`CREATE TABLE links (
|
|
from_url_id INTEGER NOT NULL,
|
|
to_url_id INTEGER NOT NULL,
|
|
FOREIGN KEY(from_url_id) REFERENCES urls(id) ON DELETE CASCADE,
|
|
FOREIGN KEY(to_url_id) REFERENCES urls(id) ON DELETE CASCADE,
|
|
PRIMARY KEY(from_url_id, to_url_id)
|
|
)`,
|
|
|
|
`CREATE INDEX links_outgoing_idx ON links (to_url_id)`,
|
|
},
|
|
},
|
|
}}
|
|
|
|
/*
|
|
TODO
|
|
- initialization options
|
|
- cleanup period
|
|
- document SQLiteStore properly
|
|
- teardown the cleanup goroutine
|
|
*/
|
|
|
|
type SQLiteStore struct {
|
|
db *sql.DB
|
|
}
|
|
|
|
var _ Store = (*SQLiteStore)(nil)
|
|
|
|
// NewInMemStore returns a Store implementation which uses an in-memory SQLite
|
|
// db.
|
|
func NewInMemStore() *SQLiteStore {
|
|
db, err := sql.Open("sqlite3", ":memory:?_foreign_keys=1")
|
|
if err != nil {
|
|
panic(fmt.Errorf("opening sqlite in memory: %w", err))
|
|
}
|
|
|
|
if _, err := migrate.Exec(db, "sqlite3", migrations, migrate.Up); err != nil {
|
|
panic(fmt.Errorf("running migrations: %w", err))
|
|
}
|
|
|
|
return &SQLiteStore{db}
|
|
}
|
|
|
|
// Close cleans up all resources held by the SQLiteStore store, if any. It must
|
|
// be the final method call to the SQLiteStore.
|
|
func (s *SQLiteStore) Close() error {
|
|
return s.db.Close()
|
|
}
|
|
|
|
// GetByStatus implements the method for the Store interface.
|
|
func (s *SQLiteStore) GetByStatus(status ResourceStatus) miter.Iterator[Resource] {
|
|
const query = `
|
|
WITH
|
|
incoming(url_id, urls) AS (
|
|
SELECT
|
|
to_url_id,
|
|
GROUP_CONCAT(url, char(0))
|
|
FROM links
|
|
JOIN urls ON (urls.id = links.from_url_id)
|
|
GROUP BY to_url_id
|
|
),
|
|
outgoing(url_id, urls) AS (
|
|
SELECT
|
|
from_url_id,
|
|
GROUP_CONCAT(url, char(0))
|
|
FROM links
|
|
JOIN urls ON (urls.id = links.to_url_id)
|
|
GROUP BY from_url_id
|
|
)
|
|
SELECT
|
|
url,
|
|
status,
|
|
pinned,
|
|
last_checked,
|
|
error_string,
|
|
incoming.urls,
|
|
outgoing.urls
|
|
FROM resources
|
|
JOIN urls ON (urls.id = resources.url_id)
|
|
LEFT JOIN incoming ON (incoming.url_id = resources.url_id)
|
|
LEFT JOIN outgoing ON (outgoing.url_id = resources.url_id)
|
|
WHERE status = ?`
|
|
|
|
return miter.Lazily(func(ctx context.Context) (miter.Iterator[Resource], error) {
|
|
rows, err := s.db.QueryContext(ctx, query, status)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("executing query: %w", err)
|
|
}
|
|
|
|
return miter.FromFunc(func(ctx context.Context) (Resource, error) {
|
|
var (
|
|
r Resource
|
|
lastChecked int64
|
|
incoming, outgoing sql.NullString
|
|
)
|
|
|
|
if !rows.Next() {
|
|
return Resource{}, errors.Join(rows.Close(), miter.ErrEnd)
|
|
}
|
|
|
|
if err := rows.Scan(
|
|
&r.URL,
|
|
&r.Status,
|
|
&r.Pinned,
|
|
&lastChecked,
|
|
&r.ErrorString,
|
|
&incoming,
|
|
&outgoing,
|
|
); err != nil {
|
|
return Resource{}, errors.Join(
|
|
rows.Close(), fmt.Errorf("scanning row: %w", err),
|
|
)
|
|
}
|
|
|
|
if lastChecked != 0 {
|
|
r.LastChecked = time.Unix(lastChecked, 0).UTC()
|
|
}
|
|
|
|
if incoming.String != "" {
|
|
if r.IncomingLinkURLs, err = parseURLs(
|
|
strings.Split(incoming.String, "\x00"),
|
|
); err != nil {
|
|
return Resource{}, errors.Join(
|
|
rows.Close(), fmt.Errorf("parsing incoming links: %w", err),
|
|
)
|
|
}
|
|
}
|
|
|
|
if outgoing.String != "" {
|
|
if r.OutgoingLinkURLs, err = parseURLs(
|
|
strings.Split(outgoing.String, "\x00"),
|
|
); err != nil {
|
|
return Resource{}, errors.Join(
|
|
rows.Close(), fmt.Errorf("parsing outgoing links: %w", err),
|
|
)
|
|
}
|
|
}
|
|
|
|
return r, nil
|
|
}), nil
|
|
})
|
|
}
|
|
|
|
// GetURLsByLastChecked implements the method for the Store interface.
|
|
func (s *SQLiteStore) GetURLsByLastChecked(
|
|
olderThan time.Time,
|
|
) miter.Iterator[URL] {
|
|
const query = `
|
|
SELECT url
|
|
FROM resources
|
|
JOIN urls ON (urls.id = resources.url_id)
|
|
WHERE last_checked < ?`
|
|
|
|
return miter.Lazily(func(ctx context.Context) (miter.Iterator[URL], error) {
|
|
rows, err := s.db.QueryContext(ctx, query, olderThan.Unix())
|
|
if err != nil {
|
|
return nil, fmt.Errorf("executing query: %w", err)
|
|
}
|
|
|
|
return miter.FromFunc(func(ctx context.Context) (URL, error) {
|
|
if !rows.Next() {
|
|
return "", errors.Join(rows.Close(), miter.ErrEnd)
|
|
}
|
|
|
|
var urlStr string
|
|
if err := rows.Scan(&urlStr); err != nil {
|
|
return "", errors.Join(
|
|
rows.Close(), fmt.Errorf("scanning url: %w", err),
|
|
)
|
|
}
|
|
|
|
url, err := ParseURL(urlStr)
|
|
if err != nil {
|
|
return "", errors.Join(
|
|
rows.Close(),
|
|
fmt.Errorf("parsing url %q from db: %w", urlStr, err),
|
|
)
|
|
}
|
|
|
|
return url, nil
|
|
}), nil
|
|
})
|
|
}
|
|
|
|
func (s *SQLiteStore) touch(ctx context.Context, urls []URL, pinned bool) (
|
|
[]int, error,
|
|
) {
|
|
var (
|
|
urlsQueryParams = make([]any, len(urls))
|
|
resourcesQueryParams = make([]any, 0, (len(urls)*2)+1)
|
|
ids = make([]int, 0, len(urls))
|
|
)
|
|
|
|
for i := range urls {
|
|
urlsQueryParams[i] = urls[i]
|
|
}
|
|
|
|
urlsQuery := `
|
|
INSERT INTO urls (url)
|
|
VALUES ` + joinRepeated("(?)", ",", len(urls)) + `
|
|
ON CONFLICT DO UPDATE SET url=url
|
|
RETURNING id`
|
|
|
|
rows, err := s.db.QueryContext(ctx, urlsQuery, urlsQueryParams...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("inserting into urls: %w", err)
|
|
}
|
|
|
|
for range urls {
|
|
if !rows.Next() {
|
|
return nil, errors.Join(
|
|
errors.New("expected a returned row"), rows.Close(),
|
|
)
|
|
}
|
|
|
|
var id int
|
|
if err := rows.Scan(&id); err != nil {
|
|
return nil, errors.Join(
|
|
fmt.Errorf("scanning return from insert into urls: %w", err),
|
|
rows.Close(),
|
|
)
|
|
}
|
|
|
|
resourcesQueryParams = append(resourcesQueryParams, id, pinned)
|
|
ids = append(ids, id)
|
|
}
|
|
|
|
rows.Close()
|
|
|
|
resourcesQuery := `
|
|
INSERT INTO resources (url_id, pinned)
|
|
VALUES ` + joinRepeated("(?,?)", ",", len(urls)) + `
|
|
ON CONFLICT DO UPDATE SET pinned = ?`
|
|
|
|
resourcesQueryParams = append(resourcesQueryParams, pinned)
|
|
|
|
_, err = s.db.ExecContext(ctx, resourcesQuery, resourcesQueryParams...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("inserting into resources: %w", err)
|
|
}
|
|
|
|
return ids, nil
|
|
}
|
|
|
|
// SetPinned implements the method for the Store interface.
|
|
func (s *SQLiteStore) SetPinned(ctx context.Context, urls []URL) error {
|
|
_, err := s.db.ExecContext(ctx, `UPDATE resources SET pinned = 0`)
|
|
if err != nil {
|
|
return fmt.Errorf("unsetting pinned on all resources: %w", err)
|
|
}
|
|
|
|
if _, err := s.touch(ctx, urls, true); err != nil {
|
|
return fmt.Errorf("pinning resources: %w", err)
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
// Update implements the method for the Store interface.
|
|
func (s *SQLiteStore) Update(
|
|
ctx context.Context,
|
|
now time.Time,
|
|
url URL,
|
|
status ResourceStatus,
|
|
errorString string,
|
|
outgoing []URL,
|
|
) error {
|
|
const resourcesQuery = `
|
|
UPDATE resources
|
|
SET
|
|
status = ?,
|
|
last_checked = ?,
|
|
error_string = ?
|
|
WHERE url_id = (SELECT id FROM urls WHERE url = ?)
|
|
RETURNING url_id`
|
|
|
|
var urlID int
|
|
|
|
err := s.db.QueryRowContext(
|
|
ctx, resourcesQuery, status, now.Unix(), errorString, url,
|
|
).Scan(&urlID)
|
|
if err != nil {
|
|
return fmt.Errorf("inserting into resources: %w", err)
|
|
}
|
|
|
|
_, err = s.db.ExecContext(
|
|
ctx, `DELETE FROM links WHERE from_url_id = ?`, urlID,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("deleting from links: %w", err)
|
|
}
|
|
|
|
if len(outgoing) == 0 {
|
|
return nil
|
|
}
|
|
|
|
outgoingIDs, err := s.touch(ctx, outgoing, false)
|
|
if err != nil {
|
|
return fmt.Errorf("touching outgoing links: %w", err)
|
|
}
|
|
|
|
linksQueryParams := make([]any, 0, len(outgoingIDs)*2)
|
|
for i := range outgoingIDs {
|
|
linksQueryParams = append(linksQueryParams, urlID, outgoingIDs[i])
|
|
}
|
|
|
|
linksQuery := `
|
|
INSERT INTO links (from_url_id, to_url_id)
|
|
VALUES ` + joinRepeated("(?,?)", ",", len(outgoing)) + `
|
|
ON CONFLICT DO NOTHING`
|
|
|
|
_, err = s.db.ExecContext(ctx, linksQuery, linksQueryParams...)
|
|
if err != nil {
|
|
return fmt.Errorf("inserting into links: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *SQLiteStore) deleteOrphans(ctx context.Context) error {
|
|
const query = `
|
|
WITH orphans AS (
|
|
SELECT url_id FROM resources
|
|
LEFT JOIN links ON (links.to_url_id = resources.url_id)
|
|
WHERE pinned = 0 AND from_url_id IS NULL
|
|
)
|
|
DELETE FROM urls WHERE id IN orphans
|
|
`
|
|
|
|
for {
|
|
res, err := s.db.ExecContext(ctx, query)
|
|
if err != nil {
|
|
return fmt.Errorf("performing delete: %w", err)
|
|
} else if n, err := res.RowsAffected(); err != nil {
|
|
return fmt.Errorf("determining rows affected: %w", err)
|
|
} else if n == 0 {
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
|
|
func joinRepeated(str, sep string, n int) string {
|
|
res := strings.Repeat(str+sep, n)
|
|
return res[:len(res)-len(sep)]
|
|
}
|
|
|