<commit_before>package main

import (
	"crypto/sha512"
	"encoding/hex"
	"io"
	"strings"
	"time"

	"github.com/flynn/flynn/Godeps/_workspace/src/github.com/jackc/pgx"
	"github.com/flynn/flynn/pkg/postgres"
	"github.com/flynn/flynn/pkg/status"
)

const UniqueViolation = "23505"

func NewPostgresFilesystem(db *postgres.DB) (Filesystem, error) {
	m := postgres.NewMigrations()
	m.Add(1,
		`CREATE TABLE files (
	file_id oid PRIMARY KEY DEFAULT lo_create(0),
	name text UNIQUE NOT NULL,
	size bigint,
	type text,
	digest text,
	created_at timestamp with time zone NOT NULL DEFAULT current_timestamp
);`,
		`CREATE FUNCTION delete_file() RETURNS TRIGGER AS $$
    BEGIN
        PERFORM lo_unlink(OLD.file_id);
        RETURN NULL;
    END;
$$ LANGUAGE plpgsql;`,
		`CREATE TRIGGER delete_file
    AFTER DELETE ON files
    FOR EACH ROW EXECUTE PROCEDURE delete_file();`,
	)
	return &PostgresFilesystem{db: db}, m.Migrate(db)
}

type PostgresFilesystem struct {
	db *postgres.DB
}

func (p *PostgresFilesystem) Status() status.Status {
	if err := p.db.Exec("SELECT 1"); err != nil {
		return status.Unhealthy
	}
	return status.Healthy
}

func (p *PostgresFilesystem) List(dir string) ([]string, error) {
	rows, err := p.db.Query("SELECT substring(name FROM '^' || $1 || '/[^/]+/?') AS path FROM files WHERE name LIKE $1 || '%' GROUP BY path", strings.TrimSuffix(dir, "/"))
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var paths []string
	for rows.Next() {
		var path *string
		if err := rows.Scan(&path); err != nil {
			return nil, err
		}
		if path != nil {
			paths = append(paths, *path)
		}
	}
	return paths, rows.Err()
}

func (p *PostgresFilesystem) Put(name string, r io.Reader, typ string) error {
	tx, err := p.db.Begin()
	if err != nil {
		return err
	}

	var id pgx.Oid
create:
	err = tx.QueryRow("INSERT INTO files (name, type) VALUES ($1, $2) RETURNING file_id", name, typ).Scan(&id)
	if e, ok := err.(pgx.PgError); ok && e.Code == UniqueViolation {
		tx.Rollback()
		tx, err = p.db.Begin()
		if err != nil {
			return err
		}

		// file exists, delete it first
		err = tx.Exec("DELETE FROM files WHERE name = $1", name)
		if err != nil {
			tx.Rollback()
			return err
		}
		goto create
	}
	if err != nil {
		tx.Rollback()
		return err
	}

	lo, err := tx.LargeObjects()
	if err != nil {
		tx.Rollback()
		return err
	}
	obj, err := lo.Open(id, pgx.LargeObjectModeWrite)
	if err != nil {
		tx.Rollback()
		return err
	}

	h := sha512.New()
	size, err := io.Copy(obj, io.TeeReader(r, h))
	if err != nil {
		tx.Rollback()
		return err
	}

	digest := hex.EncodeToString(h.Sum(nil))
	err = tx.Exec("UPDATE files SET size = $2, digest = $3 WHERE file_id = $1", id, size, digest)
	if err != nil {
		tx.Rollback()
		return err
	}

	return tx.Commit()
}

func (p *PostgresFilesystem) Copy(dstPath, srcPath string) error {
	src, err := p.Open(srcPath)
	if err != nil {
		return err
	}
	defer src.Close()
	return p.Put(dstPath, src, "")
}

func (p *PostgresFilesystem) Delete(name string) error {
	// use a regular expression so that either a file with the name is
	// deleted, or any file prefixed with "{name}/" is deleted (so in other
	// words, mimic either deleting a file or recursively deleting a
	// directory)
	return p.db.Exec("DELETE FROM files WHERE name ~ ('^' || $1 || '(/.*)?$')", name)
}

func (p *PostgresFilesystem) Open(name string) (File, error) {
	tx, err := p.db.Begin()
	if err != nil {
		return nil, err
	}

	var f pgFile
	err = tx.QueryRow("SELECT file_id, size, type, digest, created_at FROM files WHERE name = $1",
		name).Scan(&f.id, &f.size, &f.typ, &f.etag, &f.mtime)
	if err != nil {
		tx.Rollback()
		if err == pgx.ErrNoRows {
			err = ErrNotFound
		}<commit_msg>blobstore: Copy file type in PostgresFilesystem<commit_after>package main

import (
	"crypto/sha512"
	"encoding/hex"
	"io"
	"strings"
	"time"

	"github.com/flynn/flynn/Godeps/_workspace/src/github.com/jackc/pgx"
	"github.com/flynn/flynn/pkg/postgres"
	"github.com/flynn/flynn/pkg/status"
)

const UniqueViolation = "23505"

func NewPostgresFilesystem(db *postgres.DB) (Filesystem, error) {
	m := postgres.NewMigrations()
	m.Add(1,
		`CREATE TABLE files (
	file_id oid PRIMARY KEY DEFAULT lo_create(0),
	name text UNIQUE NOT NULL,
	size bigint,
	type text,
	digest text,
	created_at timestamp with time zone NOT NULL DEFAULT current_timestamp
);`,
		`CREATE FUNCTION delete_file() RETURNS TRIGGER AS $$
    BEGIN
        PERFORM lo_unlink(OLD.file_id);
        RETURN NULL;
    END;
$$ LANGUAGE plpgsql;`,
		`CREATE TRIGGER delete_file
    AFTER DELETE ON files
    FOR EACH ROW EXECUTE PROCEDURE delete_file();`,
	)
	return &PostgresFilesystem{db: db}, m.Migrate(db)
}

type PostgresFilesystem struct {
	db *postgres.DB
}

func (p *PostgresFilesystem) Status() status.Status {
	if err := p.db.Exec("SELECT 1"); err != nil {
		return status.Unhealthy
	}
	return status.Healthy
}

func (p *PostgresFilesystem) List(dir string) ([]string, error) {
	rows, err := p.db.Query("SELECT substring(name FROM '^' || $1 || '/[^/]+/?') AS path FROM files WHERE name LIKE $1 || '%' GROUP BY path", strings.TrimSuffix(dir, "/"))
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var paths []string
	for rows.Next() {
		var path *string
		if err := rows.Scan(&path); err != nil {
			return nil, err
		}
		if path != nil {
			paths = append(paths, *path)
		}
	}
	return paths, rows.Err()
}

func (p *PostgresFilesystem) Put(name string, r io.Reader, typ string) error {
	tx, err := p.db.Begin()
	if err != nil {
		return err
	}

	var id pgx.Oid
create:
	err = tx.QueryRow("INSERT INTO files (name, type) VALUES ($1, $2) RETURNING file_id", name, typ).Scan(&id)
	if e, ok := err.(pgx.PgError); ok && e.Code == UniqueViolation {
		tx.Rollback()
		tx, err = p.db.Begin()
		if err != nil {
			return err
		}

		// file exists, delete it first
		err = tx.Exec("DELETE FROM files WHERE name = $1", name)
		if err != nil {
			tx.Rollback()
			return err
		}
		goto create
	}
	if err != nil {
		tx.Rollback()
		return err
	}

	lo, err := tx.LargeObjects()
	if err != nil {
		tx.Rollback()
		return err
	}
	obj, err := lo.Open(id, pgx.LargeObjectModeWrite)
	if err != nil {
		tx.Rollback()
		return err
	}

	h := sha512.New()
	size, err := io.Copy(obj, io.TeeReader(r, h))
	if err != nil {
		tx.Rollback()
		return err
	}

	digest := hex.EncodeToString(h.Sum(nil))
	err = tx.Exec("UPDATE files SET size = $2, digest = $3 WHERE file_id = $1", id, size, digest)
	if err != nil {
		tx.Rollback()
		return err
	}

	return tx.Commit()
}

func (p *PostgresFilesystem) Copy(dstPath, srcPath string) error {
	src, err := p.Open(srcPath)
	if err != nil {
		return err
	}
	defer src.Close()
	return p.Put(dstPath, src, src.(*pgFile).typ)
}

func (p *PostgresFilesystem) Delete(name string) error {
	// use a regular expression so that either a file with the name is
	// deleted, or any file prefixed with "{name}/" is deleted (so in other
	// words, mimic either deleting a file or recursively deleting a
	// directory)
	return p.db.Exec("DELETE FROM files WHERE name ~ ('^' || $1 || '(/.*)?$')", name)
}

func (p *PostgresFilesystem) Open(name string) (File, error) {
	tx, err := p.db.Begin()
	if err != nil {
		return nil, err
	}

	var f pgFile
	err = tx.QueryRow("SELECT file_id, size, type, digest, created_at FROM files WHERE name = $1",
		name).Scan(&f.id, &f.size, &f.typ, &f.etag, &f.mtime)
	if err != nil {
		tx.Rollback()
		if err == pgx.ErrNoRows {
			err = ErrNotFound
		}
{'fraction_non_alphanumeric': 0.14896182183523107, 'fraction_numerical': 0.006430006697923644, 'mean_word_length': 3.428232502965599, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4801074', 'n_tokens_mistral': 2958, 'n_tokens_neox': 2792, 'n_words': 824}