20151007094937_create_bookmarks.rb

class CreateBookmarks < ActiveRecord::Migration
  def up
    execute <<-SQL
CREATE SEQUENCE bookmarks_id_seq;

CREATE FUNCTION next_bookmark_id(OUT result bigint) AS $$
DECLARE
  epoch bigint := 1420070400000;
  seq_id bigint;
  now bigint;
  shard_id int := 0;
BEGIN
  SELECT nextval('bookmarks_id_seq') % 1024 INTO seq_id;

  SELECT FLOOR(EXTRACT(EPOCH FROM clock_timestamp()) * 1000) INTO now;
  result := (now - epoch) << 23;
  result := result | (shard_id << 10);
  result := result | (seq_id);
END;
$$ LANGUAGE plpgsql;

CREATE TABLE bookmarks (
  id bigint PRIMARY KEY NOT NULL DEFAULT next_bookmark_id(),
  bytes bytea,
  user_id bigint NOT NULL,
  event_id bigint NOT NULL
);

CREATE INDEX bookmarks_user_id ON bookmarks USING btree (user_id);
CREATE INDEX bookmarks_event_id ON bookmarks USING btree (event_id);
CREATE UNIQUE INDEX bookmarks_user_id_event_id ON bookmarks USING btree (user_id, event_id);

CREATE SEQUENCE bookmarks_stream_index_seq;

CREATE FUNCTION next_bookmark_stream_index(OUT result bigint) AS $$
DECLARE
  epoch bigint := 1420070400000;
  seq_id bigint;
  now bigint;
  shard_id int := 0;
BEGIN
  SELECT nextval('bookmarks_stream_index_seq') % 1024 INTO seq_id;

  SELECT FLOOR(EXTRACT(EPOCH FROM clock_timestamp()) * 1000) INTO now;
  result := (now - epoch) << 23;
  result := result | (shard_id << 10);
  result := result | (seq_id);
END;
$$ LANGUAGE plpgsql;

CREATE TABLE bookmarks_stream (
  index bigint PRIMARY KEY NOT NULL DEFAULT next_bookmark_stream_index(),
  id bigint NOT NULL,
  bytes bytea
);

CREATE INDEX bookmarks_stream_id ON bookmarks_stream USING btree (id);

CREATE FUNCTION propagate_bookmark_change_to_stream() RETURNS trigger AS $$
BEGIN
  IF (TG_OP != 'INSERT') THEN
    DELETE FROM bookmarks_stream WHERE id = OLD.id;
  END IF;

  IF (TG_OP = 'DELETE') THEN
    INSERT INTO bookmarks_stream (id, bytes) VALUES (OLD.id, NULL);
  ELSE
    INSERT INTO bookmarks_stream (id, bytes) VALUES (NEW.id, NEW.bytes);
  END IF;

  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER
  propagate_bookmark_change_to_stream
AFTER
  INSERT OR UPDATE OR DELETE
ON
  bookmarks
FOR
  EACH ROW
EXECUTE PROCEDURE
  propagate_bookmark_change_to_stream();
SQL
  end

  def down
    execute <<-SQL
DROP TRIGGER propagate_bookmark_change_to_stream ON bookmarks;
DROP FUNCTION propagate_bookmark_change_to_stream();

DROP INDEX bookmarks_stream_id;

DROP TABLE bookmarks_stream;

DROP FUNCTION next_bookmark_stream_index();

DROP SEQUENCE bookmarks_stream_index_seq;

DROP INDEX bookmarks_user_id_event_id;
DROP INDEX bookmarks_event_id;
DROP INDEX bookmarks_user_id;

DROP TABLE bookmarks;

DROP FUNCTION next_bookmark_id();

DROP SEQUENCE bookmarks_id_seq;
SQL
  end
end