FIX: under some conditions draft would say it was saving when not

This is a major change to draft internals. Previously there were quite a
few cases where the draft system would say "draft saved", when in fact
we just skipped saving.

This commit ensures the draft system deals with draft ownership handover in
a predictable way.

For example:

- Window 1 editing draft
- Window 2 editing same draft at the same time

Previously we would allow window 1 and 2 to just fight on the same draft
each window overwriting the same draft over an over.

This commit introduces an ownership concept where either window 1 or 2 win
and user is prompted on the loser window to reload screen to correct the issue

This also corrects edge cases where a user could have multiple browser windows
open and posts in 1 window, later to post in the second window. Previously
drafts would break in the second window, this corrects it.

app/models/draft.rb

@@ -5,44 +5,122 @@ class Draft < ActiveRecord::Base
   NEW_PRIVATE_MESSAGE ||= 'new_private_message'
   EXISTING_TOPIC ||= 'topic_'
 
-  def self.set(user, key, sequence, data)
+  class OutOfSequence < StandardError; end
+
+  def self.set(user, key, sequence, data, owner = nil)
     if SiteSetting.backup_drafts_to_pm_length > 0 && SiteSetting.backup_drafts_to_pm_length < data.length
       backup_draft(user, key, sequence, data)
     end
 
-    if d = find_draft(user, key)
-      return if d.sequence > sequence
+    # this is called a lot so we should micro optimize here
+    draft_id, current_owner, current_sequence = DB.query_single(<<~SQL, user_id: user.id, key: key)
+      WITH draft AS (
+        SELECT id, owner FROM drafts
+        WHERE
+          user_id = :user_id AND
+          draft_key = :key
+      ),
+      draft_sequence AS (
+        SELECT sequence
+        FROM draft_sequences
+        WHERE
+          user_id = :user_id AND
+          draft_key = :key
+      )
 
-      DB.exec(<<~SQL, id: d.id, sequence: sequence, data: data)
+      SELECT
+        (SELECT id FROM draft),
+        (SELECT owner FROM draft),
+        (SELECT sequence FROM draft_sequence)
+    SQL
+
+    current_sequence ||= 0
+
+    if draft_id
+      if current_sequence != sequence
+        raise Draft::OutOfSequence
+      end
+
+      if owner && current_owner && current_owner != owner
+        sequence += 1
+
+        DraftSequence.upsert({
+            sequence: sequence,
+            draft_key: key,
+            user_id: user.id,
+          },
+          unique_by: [:user_id, :draft_key]
+        )
+      end
+
+      DB.exec(<<~SQL, id: draft_id, sequence: sequence, data: data, owner: owner || current_owner)
         UPDATE drafts
            SET sequence = :sequence
              , data = :data
              , revisions = revisions + 1
+             , owner = :owner
          WHERE id = :id
       SQL
+
+    elsif sequence != current_sequence
+      raise Draft::OutOfSequence
     else
-      Draft.create!(user_id: user.id, draft_key: key, data: data, sequence: sequence)
+      Draft.create!(
+        user_id: user.id,
+        draft_key: key,
+        data: data,
+        sequence: sequence,
+        owner: owner
+      )
     end
 
-    true
+    sequence
   end
 
   def self.get(user, key, sequence)
-    d = find_draft(user, key)
-    d.data if d && d.sequence == sequence
+
+    opts = {
+      user_id: user.id,
+      draft_key: key,
+      sequence: sequence
+    }
+
+    current_sequence, data, draft_sequence = DB.query_single(<<~SQL, opts)
+      WITH draft AS (
+        SELECT data, sequence
+        FROM drafts
+        WHERE draft_key = :draft_key AND user_id = :user_id
+      ),
+      draft_sequence AS (
+        SELECT sequence
+        FROM draft_sequences
+        WHERE draft_key = :draft_key AND user_id = :user_id
+      )
+      SELECT
+        ( SELECT sequence FROM draft_sequence) ,
+        ( SELECT data FROM draft ),
+        ( SELECT sequence FROM draft )
+    SQL
+
+    current_sequence ||= 0
+
+    if sequence != current_sequence
+      raise Draft::OutOfSequence
+    end
+
+    data if current_sequence == draft_sequence
   end
 
   def self.clear(user, key, sequence)
-    d = find_draft(user, key)
-    d.destroy if d && d.sequence <= sequence
-  end
+    current_sequence = DraftSequence.current(user, key)
 
-  def self.find_draft(user, key)
-    if user.is_a?(User)
-      find_by(user_id: user.id, draft_key: key)
-    else
-      find_by(user_id: user, draft_key: key)
+    # bad caller is a reason to complain
+    if sequence != current_sequence
+      raise Draft::OutOfSequence
     end
+
+    # corrupt data is not a reason not to leave data
+    Draft.where(user_id: user.id, draft_key: key).destroy_all
   end
 
   def self.stream(opts = nil)